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

// #define EVTDEBUG

#define NUMSOCK   1          //set to 7 for old configuration
#define MAXREAD  65536       //64kB wiznet buffer

#include <stdlib.h>
#include <stdint.h>
#include <stdarg.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
};

#define MIN_LEN  32             // min #bytes needed to interpret FADheader
#define MAX_LEN 256*1024        // size of read-buffer per socket

//#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 factOut (int severity, int err, char *message);
extern void factReportIncomplete (uint64_t rep);

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

extern int subProcEvt (int threadID, PEVNT_HEADER * fadhd, EVENT * event,
                       int8_t * 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 evtCtrl_frstPtr;   // First event in queue
int evtCtrl_lastPtr;   // pointer to next free slot

int g_maxProc;
int gi_maxProc;

uint g_actTime;
int g_runStat;
int g_reset;

size_t g_maxMem;                //maximum memory allowed for buffer

FACT_SOCK g_port[NBOARDS];      // .addr=string of IP-addr in dotted-decimal "ddd.ddd.ddd.ddd"

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

//EVT_STAT gi;
GUI_STAT gj;

#define MAX_EVT   65536  // ( 300s @ 220Hz; 16GB = 5000 evt @ roi=1024 (27s) ; 18000 evt @ roi = 300 )
#define MAX_RUN       8  // Number of concurrent runs

EVT_CTRL evtCtrl[MAX_EVT]; //control of events during processing

void factPrintf(int severity, int id, const char *fmt, ...)
{
    char str[1000];

    va_list ap;
    va_start(ap, fmt);
    vsnprintf(str, 1000, fmt, ap);
    va_end(ap);

    factOut(severity, id, str);
}


#define MAX_HEAD_MEM (NBOARDS * sizeof(PEVNT_HEADER))
#define MAX_TOT_MEM (sizeof(EVENT) + (NPIX+NTMARK)*1024*2 + MAX_HEAD_MEM)
typedef struct TGB_struct
{
    struct TGB_struct *prev;
    void *mem;
} TGB_entry;

TGB_entry  *tgb_last = NULL;
uint64_t    tgb_memory = 0;
uint64_t    tgb_inuse  = 0;

void *TGB_Malloc()
{
    // No free slot available, next alloc would exceed max memory
    if (!tgb_last && tgb_memory+MAX_TOT_MEM>g_maxMem)
        return NULL;

    // We will return this amount of memory
    tgb_inuse += MAX_TOT_MEM;
    gj.bufTot++;

    // No free slot available, allocate a new one
    if (!tgb_last)
    {
        tgb_memory += MAX_TOT_MEM;
        return malloc(MAX_TOT_MEM);
    }

    // Get the next free slot from the stack and return it
    TGB_entry *last = tgb_last;

    void *mem = last->mem;
    tgb_last  = last->prev;

    free(last);

    return mem;
};

void TGB_free(void *mem)
{
    if (!mem)
        return;

    // Add the last free slot to the stack
    TGB_entry *entry = (TGB_entry*)malloc(sizeof(TGB_entry));

    // FIXME: Really free memory if memory usuage exceeds g_maxMem

    entry->prev = tgb_last;
    entry->mem  = mem;

    tgb_last = entry;

    // Decrease the amont of memory in use accordingly
    tgb_inuse -= MAX_TOT_MEM;
}

RUN_CTRL runCtrl[MAX_RUN];

/*
*** Definition of rdBuffer to read in IP packets; keep it global !!!!
 */

typedef union
{
   uint8_t B[MAX_LEN];
   uint16_t S[MAX_LEN / 2];
   uint32_t I[MAX_LEN / 4];
   uint64_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
   int32_t skip;                //number of bytes skipped before start of event

   int errCnt;                  //how often connect failed since last successful
   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 Port;

   CNV_FACT *rBuf;

} READ_STRUCT;

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


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


int
GenSock (int flag, int sid, int port, struct sockaddr_in *sockAddr,
         READ_STRUCT * rs)
{
/*
*** 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
 */

   //close socket if open
   if (rs->sockStat == 0)
   {
      if (close (rs->socket) > 0) {
          factPrintf(kFatal, 771, "Closing socket %d failed: %m (close,rc=%d)", sid, errno);
      } else {
          factPrintf(kInfo, 771, "Succesfully closed socket %d", sid);
      }
   }

   rs->sockStat = 99;

   if (flag < 0) {
      free (rs->rBuf);          //and never open again
      rs->rBuf = NULL;
      return 0;
   }


   if (flag == 0) {             //generate address and buffer ...
      rs->Port = port;
      rs->SockAddr.sin_family = sockAddr->sin_family;
      rs->SockAddr.sin_port = htons (port);
      rs->SockAddr.sin_addr = sockAddr->sin_addr;

      rs->rBuf = (CNV_FACT*)malloc (sizeof (CNV_FACT));
      if (rs->rBuf == NULL) {
         factPrintf(kFatal, 774, "Could not create local buffer %d (malloc failed)", sid);
         rs->sockStat = 77;
         return -3;
      }
   }


   if ((rs->socket = socket (PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0) {
       factPrintf(kFatal, 773, "Generating socket %d failed: %m (socket,rc=%d)", sid, errno);
      rs->sockStat = 88;
      return -2;
   }

   int optval = 1;
   if (setsockopt (rs->socket, SOL_SOCKET, SO_KEEPALIVE, &optval, sizeof(int)) < 0) {
       factPrintf(kInfo, 173, "Setting SO_KEEPALIVE for socket %d failed: %m (setsockopt,rc=%d)", sid, errno);
   }
   optval = 10;                 //start after 10 seconds
   if (setsockopt (rs->socket, SOL_TCP, TCP_KEEPIDLE, &optval, sizeof(int)) < 0) {
       factPrintf(kInfo, 173, "Setting TCP_KEEPIDLE for socket %d failed: %m (setsockopt,rc=%d)", sid, errno);
   }
   optval = 10;                 //do every 10 seconds
   if (setsockopt (rs->socket, SOL_TCP, TCP_KEEPINTVL, &optval, sizeof(int)) < 0) {
      factPrintf(kInfo, 173, "Setting TCP_KEEPINTVL for socket %d failed: %m (setsockopt,rc=%d)", sid, errno);
   }
   optval = 2;                  //close after 2 unsuccessful tries
   if (setsockopt (rs->socket, SOL_TCP, TCP_KEEPCNT, &optval, sizeof(int)) < 0) {
      factPrintf(kInfo, 173, "Setting TCP_KEEPCNT for socket %d failed: %m (setsockopt,rc=%d)", sid, errno);
   }

   factPrintf(kInfo, 773, "Successfully generated socket %d", sid);

   rs->sockStat = -1;           //try to (re)open socket
   rs->errCnt = 0;
   return 0;

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

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

int checkRoiConsistency(const CNV_FACT *rbuf, int roi[])
{
    int xjr = -1;
    int xkr = -1;

    //points to the very first roi
    int roiPtr = sizeof(PEVNT_HEADER)/2 + 2;

    roi[0] = ntohs(rbuf->S[roiPtr]);

    for (int jr = 0; jr < 9; jr++)
    {
        roi[jr] = ntohs(rbuf->S[roiPtr]);

        if (roi[jr]<0 || roi[jr]>1024)
        {
            factPrintf(kError, 999, "Illegal roi in channel %d (allowed: 0<=roi<=1024)", jr, roi[jr]);
            return 0;
        }

        // Check that the roi of pixels jr are compatible with the one of pixel 0
        if (jr!=8 && roi[jr]!=roi[0])
        {
            xjr = jr;
            break;
        }

        // Check that the roi of all other DRS chips on boards are compatible
        for (int kr = 1; kr < 4; kr++)
        {
            const int kroi = ntohs(rbuf->S[roiPtr]);
            if (kroi != roi[jr])
            {
                xjr = jr;
                xkr = kr;
                break;
            }
            roiPtr += kroi+4;
        }
    }

    if (xjr>=0)
    {
        if (xkr<0)
            factPrintf(kFatal, 1, "Inconsistent Roi accross chips [DRS=%d], expected %d, got %d", xjr, roi[0], roi[xjr]);
        else
            factPrintf(kFatal, 1, "Inconsistent Roi accross channels [DRS=%d Ch=%d], expected %d, got %d", xjr, xkr, roi[xjr], ntohs(rbuf->S[roiPtr]));

        return 0;
    }

    if (roi[8] < roi[0])
    {
        factPrintf(kError, 712, "Mismatch of roi (%d) in channel 8. Should be larger or equal than the roi (%d) in channel 0.", roi[8], roi[0]);
        //gj.badRoiB++;
        //gj.badRoi[b]++;
        return 0;
    }

    return 1;
}

int mBufEvt(const READ_STRUCT *rs)
{
    int nRoi[9];
    if (!checkRoiConsistency(rs->rBuf, nRoi))
        return -9999;

    const int  evID   = rs->evtID;
    const uint runID  = rs->runID;
    const int  trgTyp = rs->ftmTyp;
    const int  trgNum = rs->ftmID;
    const int  fadNum = rs->evtID;

    const int beg = (evtCtrl_lastPtr + MAX_EVT - 1) % MAX_EVT;
    const int end = (evtCtrl_frstPtr + MAX_EVT - 1) % MAX_EVT;

    for (int k=beg; k!=end; k=(k+MAX_EVT-1)%MAX_EVT)
    {
        // If the run is different, go on searching.
        // We cannot stop searching if a lower run-id is found as in
        // the case of the events, because theoretically, there
        // can be the same run on two different days.
        if (runID != evtCtrl[k].runNum)
            continue;

        // If the ID of the new event if higher than the last one stored
        // in that run, we have to assign a new slot (leave the loop)
        if (evID > evtCtrl[k].evNum/* && runID == evtCtrl[k].runNum*/)
            break;

        if (evID != evtCtrl[k].evNum/* || runID != evtCtrl[k].runNum*/)
            continue;

        // We have found an entry with the same runID and evtID
        // Check if ROI is consistent
        if (evtCtrl[k].nRoi != nRoi[0] || evtCtrl[k].nRoiTM != nRoi[8])
        {
            factPrintf(kError, 821, "Mismatch of roi within event. Expected roi=%d and roi_tm=%d, got %d and %d.",
                       evtCtrl[k].nRoi, evtCtrl[k].nRoiTM, nRoi[0], nRoi[8]);
            return -8201;
        }

        // count for inconsistencies
        if (evtCtrl[k].trgNum != trgNum)
            evtCtrl[k].Errors[0]++;
        if (evtCtrl[k].fadNum != fadNum)
            evtCtrl[k].Errors[1]++;
        if (evtCtrl[k].trgTyp != trgTyp)
            evtCtrl[k].Errors[2]++;

        //everything seems fine so far ==> use this slot ....
        return k;
    }

    // Check if the control structure still has space left
    if (end-beg==1 || (end==0 && beg==MAX_EVT-1))
    {
        factPrintf(kError, 881, "No control slot to keep event %d (run %d) %d %d", evID, runID, beg, end);
        return -1;
    }

    // If we have already queued at least one event,
    // check the roi if the previous event
    // already belongs to the same run.

    // Get the runCtrl entry of the previous event.
    // If none is in the queue (we are anyhow super fast)
    // just get the correct entry from the runCtrl array.
    int idx = evtCtrl[beg].runCtrl_idx;

    // If there is an event in the queue and it has the same runID, we can use
    // that event to check for the roi consistency throughout the run
    if (evtCtrl_frstPtr!=evtCtrl_lastPtr && runCtrl[idx].runId==runID)
    {
        // Check if run already registered (old entries should have runId==-1)
        if (runCtrl[idx].roi0 != nRoi[0] || runCtrl[idx].roi8 != nRoi[8])
        {
            factPrintf(kError, 931, "Mismatch of roi within run. Expected roi=%d and roi_tm=%d, got %d and %d (runID=%d, evID=%d)",
                       runCtrl[idx].roi0, runCtrl[idx].roi8, nRoi[0], nRoi[8], runID, evID);
            return -9301;
        }
    }

    // If there is none in the queue, we have to search for the correct entry
    if (evtCtrl_frstPtr==evtCtrl_lastPtr)
    {
        idx = -1;

        for (int k=0; k<MAX_RUN; k++)
        {
            if (runCtrl[k].runId==runID)
            {
                idx = k;
                break;
            }
        }
    }

    struct timeval tv;
    gettimeofday(&tv, NULL);

    const uint32_t tsec = tv.tv_sec;
    const uint32_t tusec = tv.tv_usec;

    // Run not yet registered, register run
    // If we haven't found a corresponding entry in the queue, or the runId has changed
    // we find the oldest empty entry in the runCtrl array and create a new runCtrl entry
    if (idx<0 || runCtrl[idx].runId!=runID)
    {
        // If there is none in the queue or
        idx = -1;

        uint oldest = g_actTime + 1000;
        for (int k=0; k<MAX_RUN; k++)
        {
            // This is just for sanity. We use the oldest free entry (until
            // we have understood the concept and can use "just" a free entry
            if (runCtrl[k].runId==0 && runCtrl[k].closeTime < oldest)
            {
                oldest = runCtrl[k].closeTime;
                idx = k;
            }
        }

        if (idx<0)
        {
            factPrintf(kFatal, 883, "Not able to register the new run %d", runID);
            return -1001;
        }

        factPrintf(kInfo, 503, "New run %d (evt=%d, idx=%d) registered with roi=%d and roi_tm=%d",
                   runID, evID, idx, nRoi[0], nRoi[8]);

        runCtrl[idx].runId      = runID;
        runCtrl[idx].roi0       = nRoi[0];  // FIXME: Make obsolete!
        runCtrl[idx].roi8       = nRoi[8];  // FIXME: Make obsolete!
        runCtrl[idx].fileId     = -2;
        runCtrl[idx].lastEvt    = 1;         // Number of events partially started to read
        runCtrl[idx].actEvt     = 0;         // Number of written events (write)
        runCtrl[idx].procEvt    = 0;         // Number of successfully checked events (checkEvent)
        runCtrl[idx].maxEvt     = 999999999; // max number events allowed
        runCtrl[idx].lastTime   = tsec;      // Time when the last event was written
        runCtrl[idx].closeTime  = tsec + 3600 * 24;     //max time allowed
    }

    // The new entry which is populated is the one lastPtr is pointing to
    const int k = evtCtrl_lastPtr;

    //flag all boards as unused
    evtCtrl[k].nBoard = 0;
    for (int b=0; b<NBOARDS; b++)
        evtCtrl[k].board[b] = -1;

    evtCtrl[k].runCtrl_idx = idx;
    evtCtrl[k].pcTime[0] = tsec;
    evtCtrl[k].pcTime[1] = tusec;
    evtCtrl[k].nRoi      = nRoi[0];
    evtCtrl[k].nRoiTM    = nRoi[8];
    evtCtrl[k].evNum     = evID;
    evtCtrl[k].runNum    = runID;
    evtCtrl[k].fadNum    = fadNum;
    evtCtrl[k].trgNum    = trgNum;
    evtCtrl[k].trgTyp    = trgTyp;
    evtCtrl[k].Errors[0] = 0;
    evtCtrl[k].Errors[1] = 0;
    evtCtrl[k].Errors[2] = 0;
    evtCtrl[k].Errors[3] = 0;
    evtCtrl[k].fEvent    = NULL;
    evtCtrl[k].FADhead   = NULL;

    //    -1:   kInValid
    //     0:   kValid
    //  1-40:   kIncomplete
    //    90:   kIncompleteReported
    //   100:   kCompleteEventInBuffer
    //  1000+x: kToBeProcessedByThreadX
    //  5000:   kToBeWritten
    // 10000:   kToBeDeleted

    evtCtrl[k].evtStat = 0;

    // This is dangerous, because theoretically, it can result is
    // acessing invalid memory in another thread if this is split
    // in two instructions. Must be done only _after_ the contents
    // have been initialized
    evtCtrl_lastPtr = (evtCtrl_lastPtr+1) % MAX_EVT;

    return k;

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


void initEvent(int i)
{
    evtCtrl[i].fEvent = (EVENT*)((char*)evtCtrl[i].FADhead+MAX_HEAD_MEM);
    memset(evtCtrl[i].fEvent->Adc_Data, 0, (NPIX+NTMARK)*2*evtCtrl[i].nRoi);

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

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

    evtCtrl[i].fEvent->NumBoards   = 0;
    evtCtrl[i].fEvent->SoftTrig    = 0;
    evtCtrl[i].fEvent->PCTime      = evtCtrl[i].pcTime[0];
    evtCtrl[i].fEvent->PCUsec      = evtCtrl[i].pcTime[1];
    evtCtrl[i].fEvent->Roi         = evtCtrl[i].nRoi;
    evtCtrl[i].fEvent->RoiTM       = evtCtrl[i].nRoiTM;
    evtCtrl[i].fEvent->EventNum    = evtCtrl[i].evNum;
    evtCtrl[i].fEvent->TriggerNum  = evtCtrl[i].trgNum;
    evtCtrl[i].fEvent->TriggerType = evtCtrl[i].trgTyp;
}


void mBufFree (int i)
{
    TGB_free(evtCtrl[i].FADhead);

    evtCtrl[i].fEvent = NULL;
    evtCtrl[i].FADhead = NULL;

    evtCtrl[i].evNum = evtCtrl[i].nRoi = -1;
    evtCtrl[i].runNum = 0;

    gj.usdMem = tgb_inuse;

    gj.bufTot--;
}

uint64_t reportIncomplete(int id, const char *txt)
{
    factPrintf(kWarn, 601, "skip incomplete evt (run=%d, evt=%d, %s)",
               evtCtrl[id].runNum, evtCtrl[id].evNum, txt);

    uint64_t report = 0;

    char str[1000];

    int ik=0;
    for (int ib=0; ib<NBOARDS; ib++)
    {
        if (ib%10==0)
            str[ik++] = '|';

        const int jb = evtCtrl[id].board[ib];
        if (jb>=0) // data received from that board
        {
            str[ik++] = '0'+(jb%10);
            continue;
        }

        // FIXME: This is not synchronous... it reports
        // accoridng to the current connection status, not w.r.t. to the
        // one when the event was taken.
        if (gi_NumConnect[ib]<=0) // board not connected
        {
            str[ik++] = 'x';
            continue;
        }

        // data from this board lost
        str[ik++] = '.';
        report |= ((uint64_t)1)<<ib;
    }

    str[ik++] = '|';
    str[ik]   = 0;

    factOut(kWarn, 601, str);

    return report;
}

// i == board
void copyData(CNV_FACT *rbuf, int i, int evID)
{
    // swapEventHeaderBytes: End of the header. to channels now
    int eStart = 36;
    for (int ePatchesCount = 0; ePatchesCount<4*9; ePatchesCount++)
    {
        rbuf->S[eStart+0] = ntohs(rbuf->S[eStart+0]);//id
        rbuf->S[eStart+1] = ntohs(rbuf->S[eStart+1]);//start_cell
        rbuf->S[eStart+2] = ntohs(rbuf->S[eStart+2]);//roi
        rbuf->S[eStart+3] = ntohs(rbuf->S[eStart+3]);//filling

        eStart += 4+rbuf->S[eStart+2];//skip the pixel data
    }

    memcpy(&evtCtrl[evID].FADhead[i], rbuf, sizeof(PEVNT_HEADER));

    int src = sizeof(PEVNT_HEADER) / 2;

    // consistency of ROIs have been checked already (is it all correct?)
    const int roi = rbuf->S[src+2];

    // different sort in FAD board.....
    for (int px = 0; px < 9; px++)
    {
        for (int drs = 0; drs < 4; drs++)
        {
            // pixH = rd[i].rBuf->S[src++];    // ID
            src++;

            const int pixC = rbuf->S[src++];    // start-cell
            const int pixR = rbuf->S[src++];    // roi
            //here we should check if pixH is correct ....

            const int pixS = i * 36 + drs * 9 + px;
            src++;

            evtCtrl[evID].fEvent->StartPix[pixS] = pixC;

            const int dest1 = pixS * roi;
            memcpy(&evtCtrl[evID].fEvent->Adc_Data[dest1], &rbuf->S[src], roi * 2);

            src += pixR;

            if (px == 8)
            {
                const int tmS = i * 4 + drs;

                //and we have additional TM info
                if (pixR > roi)
                {
                    const int dest2 = tmS * roi + NPIX * roi;

                    const int srcT = src - roi;
                    evtCtrl[evID].fEvent->StartTM[tmS] = (pixC + pixR - roi) % 1024;

                    memcpy(&evtCtrl[evID].fEvent->Adc_Data[dest2], &rbuf->S[srcT], roi * 2);
                }
                else
                {
                    evtCtrl[evID].fEvent->StartTM[tmS] = -1;
                }
            }
        }
    }
}


void *readFAD (void *ptr)
{
/* *** main loop reading FAD data and sorting them to complete events */

    factPrintf(kInfo, -1, "Start initializing (readFAD)");

    READ_STRUCT rd[NBOARDS];       //buffer to read IP and afterwards store in mBuffer

    uint32_t actrun = 0;

   const int minLen = sizeof(PEVNT_HEADER);  //min #bytes needed to check header: full header for debug


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

   int gi_reset, gi_resetR, gi_resetS, gi_resetW, gi_resetX;
   gi_resetS = gi_resetR = 9;

   int sockDef[NBOARDS];        //internal state of sockets
   memset(sockDef, 0, NBOARDS*sizeof(int));

 START:
   evtCtrl_frstPtr = 0;
   evtCtrl_lastPtr = 0;

   //time in seconds
   uint gi_SecTime = time(NULL);;
   g_actTime = gi_SecTime;

   const int cntsock = 8 - NUMSOCK ;

   if (gi_resetS > 0) {
      //make sure all sockets are preallocated as 'not exist'
      for (int i = 0; i < NBOARDS; i++) {
         rd[i].socket = -1;
         rd[i].sockStat = 99;
      }

      for (int 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)
   {
      gj.bufTot  = gj.maxEvt = gj.xxxEvt = 0;
      gj.usdMem  = gj.maxMem = gj.xxxMem = 0;
      gj.totMem  = tgb_memory;
      gj.bufNew  = gj.bufEvt = 0;
      gj.evtSkip = gj.evtWrite = gj.evtErr = 0;

      // initialize mBuffer (mark all entries as unused\empty)
      for (int i = 0; i < MAX_EVT; i++)
      {
          evtCtrl[i].evNum = evtCtrl[i].nRoi = -1;
          evtCtrl[i].runNum = 0;
          evtCtrl[i].evtStat = -1;
      }

      evtCtrl_frstPtr = 0;
      evtCtrl_lastPtr = 0;

      factPrintf(kInfo, -1, "End   initializing (readFAD)");
   }


   gi_reset = gi_resetR = gi_resetS = gi_resetW = 0;

   //loop until global variable g_runStat claims stop
   while (g_runStat >= 0 && g_reset == 0)
   {
      gj.readStat = g_runStat;

      for (int b = 0; b < NBOARDS; b++)
      {
          // Nothing has changed
          if (g_port[b].sockDef == sockDef[b])
              continue;

          gi_NumConnect[b] = 0;       //must close all connections
          gj.numConn[b] = 0;

          // s0 =  0: sockets to be defined and opened
          // s0 = -1: sockets to be destroyed
          // s0 = +1: sockets to be closed and reopened

          int s0 = 0;
          if (sockDef[b] != 0)
              s0 = g_port[b].sockDef==0 ? -1 : +1;

          const int p0 = s0==0 ? ntohs (g_port[b].sockAddr.sin_port) : 0;

          GenSock(s0, b, p0+1, &g_port[b].sockAddr, &rd[b]); //generate address and socket

          sockDef[b] = g_port[b].sockDef;
      }

      // count the number of active boards
      int actBoards = 0;
      for (int b = 0; b < NBOARDS; b++)
          if (sockDef[b] > 0)
              actBoards++;

      //check all sockets if something to read
      for (int i = 0; i < NBOARDS; i++)
      {
          // Do not try to connect this socket
          if (rd[i].sockStat > 0)
              continue;

          if (rd[i].sockStat == -1)
          {
              //try to connect if not yet done
              rd[i].sockStat = connect (rd[i].socket,
                                        (struct sockaddr *) &rd[i].SockAddr,
                                        sizeof (rd[i].SockAddr));
              // Failed
              if (rd[i].sockStat == -1)
              {
                  rd[i].errCnt++;
                  usleep(25000);
                  continue;
              }

              // Success (rd[i].sockStat == 0)

              if (sockDef[i] > 0)
              {
                  rd[i].bufTyp = 0;                     // expect a header
                  rd[i].bufLen = sizeof(PEVNT_HEADER);  // 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[i] += cntsock;
              gj.numConn[i]++;

              factPrintf(kInfo, -1, "New connection %d (number of connections: %d)", i, gj.numConn[i]);
          }

          // Do not read from this socket
          if (rd[i].bufLen<0)
              continue;

          if (rd[i].bufLen>0)
          {
              const int32_t jrd =
                  recv(rd[i].socket, &rd[i].rBuf->B[rd[i].bufPos],
                       rd[i].bufLen, MSG_DONTWAIT);

              // recv failed
              if (jrd<0)
              {
                  // There was just nothing waiting
                  if (errno==EWOULDBLOCK || errno==EAGAIN)
                      continue;

                  factPrintf(kError, 442, "Reading from socket %d failed: %m (recv,rc=%d)", i, errno);
                  continue;
              }

              // connection was closed ...
              if (jrd==0)
              {
                  factPrintf(kInfo, 441, "Socket %d closed by FAD", i);

                  const int s0 = sockDef[i] > 0 ? +1 : -1;
                  GenSock(s0, i, 0, NULL, &rd[i]);

                  gi_NumConnect[i]-= cntsock ;
                  gj.numConn[i]--;

                  continue;
              }

              gj.rateBytes[i] += jrd;

              // are we skipping this board ...
              if (rd[i].bufTyp < 0)
                  continue;

              rd[i].bufPos += jrd;  //==> prepare for continuation
              rd[i].bufLen -= jrd;

#ifdef EVTDEBUG
              debugRead(i, jrd, rd[i].evtID, rd[i].ftmID, rd[i].runID, rd[i].bufTyp, tv.tv_sec, tv.tv_usec); // i=socket; jrd=#bytes; ievt=eventid; 1=finished event
#endif
          }

          //we are reading event header
          if (rd[i].bufTyp <= 0)
          {
              //not yet sufficient data to take action
              if (rd[i].bufPos < minLen)
                  continue;

              //check if startflag correct; else shift block ....
              // FIXME: This is not enough... this combination of
              //        bytes can be anywhere... at least the end bytes
              //        must be checked somewhere, too.
              int k;
              for (k = 0; k < rd[i].bufPos - 1; k++)
              {
                  //start.S = 0xFB01;
                  if (rd[i].rBuf->B[k] == 0xfb && rd[i].rBuf->B[k+1] == 0x01)
                      break;
              }
              rd[i].skip += k;

              //no start of header found
              if (k >= rd[i].bufPos - 1)
              {
                  rd[i].rBuf->B[0] = rd[i].rBuf->B[rd[i].bufPos];
                  rd[i].bufPos = 1;
                  rd[i].bufLen = sizeof(PEVNT_HEADER)-1;
                  continue;
              }

              if (k > 0)
              {
                  rd[i].bufPos -= k;
                  rd[i].bufLen += k;
                  memmove (&rd[i].rBuf->B[0], &rd[i].rBuf->B[k],
                           rd[i].bufPos);
              }

              if (rd[i].bufPos < minLen)
                  continue;

              if (rd[i].skip > 0)
              {
                  factPrintf(kInfo, 666, "Skipped %d bytes on port %d", rd[i].skip, i);
                  rd[i].skip = 0;
              }

              // TGB: This needs much more checks than just the first two bytes!

              // Swap everything except start_package_flag.
              // It is to difficult to find out where it is used how,
              // but it doesn't really matter because it is not really
              // used anywehere else
              //   rd[i].rBuf->S[1]  = ntohs(rd[i].rBuf->S[1]);    // package_length
              rd[i].rBuf->S[2]  = ntohs(rd[i].rBuf->S[2]);    // version_no
              rd[i].rBuf->S[3]  = ntohs(rd[i].rBuf->S[3]);    // PLLLCK
              rd[i].rBuf->S[4]  = ntohs(rd[i].rBuf->S[4]);    // trigger_crc
              rd[i].rBuf->S[5]  = ntohs(rd[i].rBuf->S[5]);    // trigger_type

              rd[i].rBuf->S[12] = ntohs(rd[i].rBuf->S[12]);   // board id
              rd[i].rBuf->S[13] = ntohs(rd[i].rBuf->S[13]);   // adc_clock_phase_shift
              rd[i].rBuf->S[14] = ntohs(rd[i].rBuf->S[14]);   // number_of_triggers_to_generate
              rd[i].rBuf->S[15] = ntohs(rd[i].rBuf->S[15]);   // trigger_generator_prescaler

              rd[i].rBuf->I[3]  = ntohl(rd[i].rBuf->I[3]);    // trigger_id
              rd[i].rBuf->I[4]  = ntohl(rd[i].rBuf->I[4]);    // fad_evt_counter
              rd[i].rBuf->I[5]  = ntohl(rd[i].rBuf->I[5]);    // REFCLK_frequency

              rd[i].rBuf->I[10] = ntohl(rd[i].rBuf->I[10]);   // runnumber;
              rd[i].rBuf->I[11] = ntohl(rd[i].rBuf->I[11]);   // time;

              for (int s=24;s<24+NTemp+NDAC;s++)
                  rd[i].rBuf->S[s] = ntohs(rd[i].rBuf->S[s]); // drs_temperature / dac

              rd[i].fadLen  = ntohs(rd[i].rBuf->S[1]) * 2;
              rd[i].fadVers = rd[i].rBuf->S[2];
              rd[i].ftmTyp  = rd[i].rBuf->S[5];
              rd[i].ftmID   = rd[i].rBuf->I[3];    //(FTMevt)
              rd[i].evtID   = rd[i].rBuf->I[4];    //(FADevt)
              rd[i].runID   = rd[i].rBuf->I[11]==0 ? g_actTime : rd[i].rBuf->I[11];
              rd[i].bufTyp  = 1;  //ready to read full record
              rd[i].bufLen  = rd[i].fadLen - rd[i].bufPos;

              const int fadBoard = rd[i].rBuf->S[12];
              debugHead(i, fadBoard, rd[i].rBuf);

              continue;
          }

          // are we reading data

          // not yet all read
          if (rd[i].bufLen > 0)
              continue;

          // stop.S = 0x04FE;
          if (rd[i].rBuf->B[rd[i].fadLen - 1] != 0xfe ||
              rd[i].rBuf->B[rd[i].fadLen - 2] != 0x04)
          {
              factPrintf(kError, 301, "End-of-event flag wrong on socket %3d for event %4d (len=%5d), got %3d %3d",
                         i, rd[i].evtID, rd[i].fadLen,
                         rd[i].rBuf->B[rd[i].fadLen - 1], rd[i].rBuf->B[rd[i].fadLen - 2]);

              // ready to read next header
              rd[i].bufTyp = 0;
              rd[i].bufLen = sizeof(PEVNT_HEADER);
              rd[i].bufPos = 0;

              continue;
          }

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

          //get index into mBuffer for this event (create if needed)
          const int idx = mBufEvt(&rd[i]);

          // no free entry in mBuffer, retry later
          if (idx == -1)
              continue;

          // We have a valid entry, but no memory has yet been allocated
          if (idx >= 0 && evtCtrl[idx].evtStat==0 && evtCtrl[idx].FADhead==NULL)
          {
              // Try to get memory from the big buffer
              evtCtrl[idx].FADhead = (PEVNT_HEADER*)TGB_Malloc();
              if (evtCtrl[idx].FADhead == NULL)
              {
                  // If this works properly, this is a hack which can be removed, or
                  // replaced by a signal or dim message
                  if (rd[i].bufTyp==1)
                      factPrintf(kError, 882, "No free memory left for %d (run=%d)", evtCtrl[idx].evNum, evtCtrl[idx].runNum);
                  rd[i].bufTyp = 2;
                  continue;
              }

              // Initialise contents of mBuffer[evID]->fEvent
              initEvent(idx);

              // Some statistics
              gj.usdMem = tgb_inuse;

              if (gj.usdMem > gj.maxMem)
                  gj.maxMem = gj.usdMem;

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

          // ready to read next header
          rd[i].bufTyp = 0;
          rd[i].bufLen = sizeof(PEVNT_HEADER);
          rd[i].bufPos = 0;

          // Fatal error occured. Event cannot be processed. Skip it. Start reading next header.
          if (idx < -1000)
              continue;

          if (evtCtrl[idx].evtStat==-1 || evtCtrl[idx].evtStat>=90)
          {
              factPrintf(kError, 882, "Received data of event %d [%d] (run=%d) has already been advanced (stat=%d)... skipping", evtCtrl[idx].evNum, i, evtCtrl[idx].runNum, evtCtrl[idx].evtStat);
              continue;
          }

          //we have a valid entry in mBuffer[]; fill it
          const int fadBoard = rd[i].rBuf->S[12];
          const int fadCrate = fadBoard>>8;

          if (i != (fadCrate * 10 + (fadBoard&0xff)))
          {
              factPrintf(kWarn, 301, "Board ID mismatch. Expected %d, got %d (C=%d, B=%d)",
                         i, fadBoard, fadCrate, fadBoard&0xff);
          }

          if (evtCtrl[idx].board[i] != -1)
          {
              factPrintf(kWarn, 501, "Got event %5d from board %3d (i=%3d, len=%5d) twice: Starts with %3d %3d - ends with %3d %3d",
                         evtCtrl[idx].evNum, i, i, rd[i].fadLen,
                         rd[i].rBuf->B[0], rd[i].rBuf->B[1],
                         rd[i].rBuf->B[rd[i].fadLen - 2],
                         rd[i].rBuf->B[rd[i].fadLen - 1]);
              continue; // Continue reading next header
          }

          // Copy data from rd[i] to mBuffer[evID]
          copyData(rd[i].rBuf, i, idx);

          // now we have stored a new board contents into Event structure

          evtCtrl[idx].fEvent->NumBoards++;
          evtCtrl[idx].board[i] = i;
          evtCtrl[idx].nBoard++;
          evtCtrl[idx].evtStat = evtCtrl[idx].nBoard;

          // have we already reported first (partial) event of this run ???
          if (evtCtrl[idx].nBoard==1 && evtCtrl[idx].runNum != actrun)
          {
             // Signal the fadctrl that a new run has been started
              gotNewRun(evtCtrl[idx].runNum, NULL);

              factPrintf(kInfo, 1, "gotNewRun called, prev run %d, new run %d, event %d",
                         actrun, evtCtrl[idx].runNum, evtCtrl[idx].evNum);

              for (int j=0; j<MAX_RUN; j++)
              {
                  // Since we have started a new run, we know already when to close the
                  // previous run in terms of number of events
                  if (runCtrl[j].runId==actrun)
                      runCtrl[j].maxEvt = runCtrl[j].lastEvt;

                  // We got the first part of this event, so this is
                  // the number of events we expect for this run
                  if (runCtrl[j].runId==evtCtrl[idx].runNum)
                      runCtrl[j].lastEvt++;
              }

              // Change 'actrun' the the new runnumber
              actrun = evtCtrl[idx].runNum;
          }

          // event not yet complete
          if (evtCtrl[idx].nBoard < actBoards)
              continue;

          // This is a non-ideal hack to lower the probability that
          // in mBufEvt the search for correct entry in runCtrl
          // will not return a super-old entry. I don't want
          // to manipulate that in another thread.
          for (int ir=0; ir<MAX_RUN; ir++)
          {
              if (runCtrl[ir].runId != actrun && runCtrl[ir].fileId>0)
                  runCtrl[ir].runId = 0;
          }

          const int beg = (idx             + MAX_EVT - 1) % MAX_EVT;
          const int end = (evtCtrl_frstPtr + MAX_EVT - 1) % MAX_EVT;

          // we have just completed an event... so all previous events
          // must have been completed already. If they are not, there
          // is no need to wait for the timeout, because they will never
          // get completed. We can just ensure that if we check for the previous
          // event to be complete every time we receive a new complete event.
          // If we find an incomplete one, we remove all consecutive
          // incomplete ones.
          for (int k=beg; k!=end; k=(k+MAX_EVT-1)%MAX_EVT)
          {
              // We are done if we find a complete or fully processed event
              if (evtCtrl[k].evtStat>=100)
                  break;

              // we do not call factReportIncomplete here, because by starting a new
              // run and having received the first complete event from that run
              // the user has expressed that the old events are obsolste now
              // and the run will be closed anyway
              if (evtCtrl[k].evtStat>0 && evtCtrl[k].evtStat<90)
              {
                  reportIncomplete(k, "expired");
                  evtCtrl[k].evtStat = 90;
              }
          }

          // Flag that the event is ready for processing
          evtCtrl[idx].evtStat = 100;

      } // end for loop over all sockets

      g_actTime = time (NULL);
      if (g_actTime <= gi_SecTime)
      {
          usleep(1);
          continue;
      }
      gi_SecTime = g_actTime;

      gj.bufNew = 0;

      //loop over all active events and flag those older than read-timeout
      //delete those that are written to disk ....

      const int count = (evtCtrl_lastPtr-evtCtrl_frstPtr+MAX_EVT)%MAX_EVT;

      // This could be improved having the pointer which separates the queue with
      // the incomplete events from the queue with the complete events
      for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
      {
          // Check the more likely case first: incomplete events
          if (evtCtrl[k0].evtStat>0 && evtCtrl[k0].evtStat<100)
          {
              gj.bufNew++;     //incomplete event in Buffer

              // Event has not yet timed out or was reported already
              if (evtCtrl[k0].evtStat==90 || evtCtrl[k0].pcTime[0]>=g_actTime - 30)
                  continue;

              // This will result in the emission of a dim service.
              // It doesn't matter if that takes comparably long,
              // because we have to stop the run anyway.
              const uint64_t rep = reportIncomplete(k0, "timeout");
              factReportIncomplete(rep);

              //timeout for incomplete events
              evtCtrl[k0].evtStat = 90;

              gj.evtSkip++;

              continue;
          }

          // Check the less likely case: 'useless' or 'delete'
          // evtState==0 can happen if the event was initialized (some data received)
          // but the data did not make sense (e.g. inconsistent rois)
          if (evtCtrl[k0].evtStat==10000)
          {
              evtCtrl[k0].evtStat = -1;
              mBufFree(k0);   //event written--> free memory

              gj.evtWrite++;
              gj.rateWrite++;
          }

          // Remove leading invalidated slots from queue
          if (evtCtrl[evtCtrl_frstPtr].evtStat==-1 && k0==evtCtrl_frstPtr)
          {
              evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % MAX_EVT;
              continue;
          }
      }

      // The number of complete events in the buffer is the total number of
      // events in the buffer minus the number of incomplete events.
      gj.bufEvt = count - gj.bufNew;

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

      for (int ib = 0; ib < NBOARDS; ib++)
          gj.totBytes[ib] += gj.rateBytes[ib];

      gj.totMem = tgb_memory;

      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 (int b = 0; b < NBOARDS; b++)
          gj.rateBytes[b] = 0;

   } // while (g_runStat >= 0 && g_reset == 0)

   factPrintf(kInfo, -1, "Stop reading ... RESET=%d", g_reset);

   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;
       gi_resetR = g_runStat == -1 ? 1 : 7;

       gi_resetS = 7;            //close all sockets
       gi_resetW = 7;            //close all files
       gi_resetX = 0;

       //inform others we have to quit ....
       gj.readStat = -11;        //inform all that no update to happen any more
   }

   if (gi_resetS > 0)
   {
       //must close all open sockets ...
       factPrintf(kInfo, -1, "Close all sockets...");

       for (int i = 0; i < NBOARDS; i++)
       {
           if (rd[i].sockStat != 0)
               continue;

           GenSock(-1, i, 0, NULL, &rd[i]);   //close and destroy open socket

           gi_NumConnect[i]-= cntsock ;
           gj.numConn[i]--;
           sockDef[i] = 0;      //flag ro recreate the sockets ...
           rd[i].sockStat = -1; //and try to open asap
       }
   }


   if (gi_resetR > 0)
   {
       //and clear all buffers (might have to wait until all others are done)
       while (evtCtrl_frstPtr!=evtCtrl_lastPtr)
       {
           const int k0=evtCtrl_frstPtr;

           // flag incomplete events as 'read finished'
           // We cannot just detele all events, because some might currently being processed,
           // so we have to wait until the processing thread currently processing the event
           // signals that the event can be deleted. (Note, that there are currently never
           // two threads processing the same event at the same time)
           if (evtCtrl[k0].evtStat<90 || evtCtrl[k0].evtStat==10000)
           {
               evtCtrl[k0].evtStat = -1;
               mBufFree(k0);   //event written--> free memory
               evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % MAX_EVT;
           }

           usleep(1);
       }
   }

   if (gi_reset > 0)
   {
      if (gi_resetW > 0)
         CloseRunFile (0, 0, 0);        //ask all Runs to be closed

      if (gi_resetX > 0)
      {
          struct timespec xwait;
          xwait.tv_sec = gi_resetX;
          xwait.tv_nsec = 0;
          nanosleep (&xwait, NULL);
      }

      factPrintf(kInfo, -1, "Continue read Process ...");
      gi_reset = 0;
      goto START;
   }

   factPrintf(kInfo, -1, "Exit read Process...");

   factPrintf(kInfo, -1, "%ld Bytes flagged as in-use.", tgb_inuse);

   gj.readStat = -99;

   factStat (gj);
   //factStatNew (gi);

   return 0;

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


void *subProc(void *thrid)
{
    const int64_t threadID = (int64_t)thrid;

    factPrintf(kInfo, -1, "Starting sub-process-thread %ld", threadID);

    while (g_runStat > -2) //in case of 'exit' we still must process pending events
    {
        int numWait = 0;

        for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
        {
           // This is a threading issue... the evtStat might have been invalid
           // but the frstPtr is not yet updated
           if (evtCtrl[k0].evtStat==-1)
               continue;

           // If we find the first event still waiting for processing
           // there will be only unprocessed events after this one in the queue
           if (evtCtrl[k0].evtStat<1000+threadID)
           {
               numWait = 1;
               break;
           }

           // If the event was processed already, skip it
           // We could replace that to a moving pointer pointing to the first
           // non-processed event
           if (evtCtrl[k0].evtStat!=1000+threadID)
               continue;

            const int jret = subProcEvt(threadID, evtCtrl[k0].FADhead, evtCtrl[k0].fEvent, 0);

            if (jret>0 && jret<=threadID)
                factPrintf(kError, -1, "Process %ld wants to send event to process %d... not allowed.", threadID, jret);

            if (jret<=threadID)
            {
                evtCtrl[k0].evtStat = 10000;            // flag as 'to be deleted'
                continue;
            }

            if (jret>=gi_maxProc)
            {
                evtCtrl[k0].evtStat = 5000;            // flag as 'to be written'
                continue;
            }

            evtCtrl[k0].evtStat = 1000 + jret;       // flag for next proces
        }

        if (gj.readStat < -10 && numWait == 0) {  //nothing left to do
            factPrintf(kInfo, -1, "Exit subProcessing in process %ld", threadID);
            return 0;
        }

        usleep(1);
    }

    factPrintf(kInfo, -1, "Ending sub-process-thread %ld", threadID);

    return 0;
}

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


void *
procEvt (void *ptr)
{
/* *** main loop processing file, including SW-trigger */
   int status;

   factPrintf(kInfo, -1, "Starting process-thread with %d subprocesses", gi_maxProc);

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

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

   // in case of 'exit' we still must process pending events
   while (g_runStat > -2)
   {
       int numWait = 0;

       for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
       {
           // This is a threading issue... the evtStat might have been invalid
           // but the frstPtr is not yet updated
           if (evtCtrl[k0].evtStat==-1)
               continue;

           // If we find the first incomplete event which is not supposed to
           // be processed, there are only more incomplete events in the queue
           if (evtCtrl[k0].evtStat<90)
           {
               numWait = 1;
               break;
           }

           // If the event was processed already, skip it.
           // We could replace that to a moving pointer pointing to the first
           // non-processed event
           if (evtCtrl[k0].evtStat>=1000)
               continue;

           //-------- it is better to open the run already here, so call can be used to initialize
           //-------- buffers etc. needed to interprete run (e.g. DRS calibration)
           const uint32_t irun = evtCtrl[k0].runNum;
           const int32_t  ievt = evtCtrl[k0].evNum;

           const int idx = evtCtrl[k0].runCtrl_idx;
           if (runCtrl[idx].runId!=irun)
           {
               //factPrintf(kFatal, 901, "procEvt: runCtrl entry for run %d vanished (evt=%d)", irun, ievt);
               // FIXME: What is the right action? (Flag event for deletion?)
               continue;
           }

           // File not yet open
           if (runCtrl[idx].fileId < 0)
           {
               RUN_HEAD actRun;
               actRun.Version =  1;
               actRun.RunType = -1;  //to be adapted
               actRun.Nroi    = evtCtrl[k0].nRoi;      //runCtrl[lastRun].roi0;
               actRun.NroiTM  = evtCtrl[k0].nRoiTM;    //runCtrl[lastRun].roi8;
               actRun.RunTime = evtCtrl[k0].pcTime[0]; //runCtrl[lastRun].firstTime;
               actRun.RunUsec = evtCtrl[k0].pcTime[1]; //runCtrl[lastRun].firstUsec;
               actRun.NBoard  = NBOARDS;
               actRun.NPix    = NPIX;
               actRun.NTm     = NTMARK;

               memcpy(actRun.FADhead, evtCtrl[k0].FADhead, NBOARDS*sizeof(PEVNT_HEADER));

               runCtrl[idx].fileHd = runOpen(irun, &actRun, sizeof (actRun));
               if (runCtrl[idx].fileHd == NULL)
               {
                   factPrintf(kError, 502, "procEvt: Could not open new file for run %d (idx=%d, evt=%d, runOpen failed)", irun, idx, ievt);
                   runCtrl[idx].fileId = 91;
                   // FIXME: What happens to evtStat? Shell we really just try again?
                   continue;
               }

               runCtrl[idx].fileId = 0;

               factPrintf(kInfo, -1, "procEvt: Opened new file for run %d (idx=%d, evt=%d)", irun, idx, ievt);
           }

           //--------
           //--------

           //and set correct event header ; also check for consistency in event (not yet)
           evtCtrl[k0].fEvent->Errors[0]   = evtCtrl[k0].Errors[0];
           evtCtrl[k0].fEvent->Errors[1]   = evtCtrl[k0].Errors[1];
           evtCtrl[k0].fEvent->Errors[2]   = evtCtrl[k0].Errors[2];
           evtCtrl[k0].fEvent->Errors[3]   = evtCtrl[k0].Errors[3];

           for (int ib=0; ib<NBOARDS; ib++)
           {
               // board is not read
               if (evtCtrl[k0].board[ib] == -1)
               {
                   evtCtrl[k0].FADhead[ib].start_package_flag = 0;
                   evtCtrl[k0].fEvent->BoardTime[ib] = 0;
               }
               else
               {
                   evtCtrl[k0].fEvent->BoardTime[ib] = evtCtrl[k0].FADhead[ib].time;
               }
           }

           const int rc = eventCheck(evtCtrl[k0].runNum, evtCtrl[k0].FADhead, evtCtrl[k0].fEvent);
           if (rc < 0)
           {
               evtCtrl[k0].evtStat = 10000;        // flag event to be deleted
           }
           else
           {
               evtCtrl[k0].evtStat = 1000;       // flag 'start processing'
               runCtrl[idx].procEvt++;
           }
       }

       if (gj.readStat < -10 && numWait == 0) {  //nothing left to do
           factPrintf(kInfo, -1, "Exit Processing Process ...");
           gj.procStat = -22;     //==> we should exit
           return 0;
       }

       usleep(1);

       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...

   factPrintf(kInfo, -1, "Abort Processing Process ...");

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

   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 checkAndCloseRun(int j, int cond, int where)
{
    if (!cond &&
        runCtrl[j].closeTime >= g_actTime &&
        runCtrl[j].lastTime >= g_actTime - 300 &&
        runCtrl[j].maxEvt > runCtrl[j].actEvt)
        return;

    //close run for whatever reason
    int ii = 0;
    if (cond)
        ii = 1;
    if (runCtrl[j].closeTime < g_actTime)
        ii |= 2; // = 2;
    if (runCtrl[j].lastTime < g_actTime - 300)
        ii |= 4; // = 3;
    if (runCtrl[j].maxEvt <= runCtrl[j].actEvt)
        ii |= 8; // = 4;

    runCtrl[j].closeTime = g_actTime - 1;

    const int rc = runClose(runCtrl[j].fileHd, NULL, 0);//&runTail[j], sizeof(runTail[j]));
    if (rc<0)
    {
        factPrintf(kError, 503, "writeEvt-%d: Error closing run %d (runClose,rc=%d)",
                   where, runCtrl[j].runId, rc);
        runCtrl[j].fileId = 92+where*2;
    }
    else
    {
        factPrintf(kInfo, 503, "writeEvt-%d: Closed run %d (reason=%d)",
                   where, runCtrl[j].runId, ii);
        runCtrl[j].fileId = 93+where*2;
    }
}

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


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

   factPrintf(kInfo, -1, "Starting write-thread");

   while (g_runStat > -2)
   {
       //int numWrite = 0;
       int numWait  = 0;

       // Note that the current loop does not at all gurantee that
       // the events are written in the correct order.
       for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
       {
           // This is a threading issue... the evtStat might have been invalid
           // but the frstPtr is not yet updated
           if (evtCtrl[k0].evtStat==-1)
               continue;

           // If we find the first non-written event which is not supposed to
           // be written, there are only more incomplete events in the queue
           if (evtCtrl[k0].evtStat<5000)
           {
               numWait = 1;
               break;
           }

           // If the event was written already already, skip it
           // We could replace that to a moving pointer pointing to the first
           // non-processed event
           if (evtCtrl[k0].evtStat!=5000)
               continue;

           const uint32_t irun = evtCtrl[k0].runNum;

           const int idx = evtCtrl[k0].runCtrl_idx;

           if (runCtrl[idx].runId!=irun)
           {
               //factPrintf(kFatal, 901, "writeEvt: runCtrl entry for run %d vanished (evt=%d)", irun, ievt);
               // FIXME: What is the right action? (Flag event for deletion?)
               continue;
           }

           // File is open
           if (runCtrl[idx].fileId==0)
           {
               const int rc = runWrite(runCtrl[idx].fileHd, evtCtrl[k0].fEvent, 0);
               if (rc >= 0)
               {
                   // Sucessfully wrote event
                   runCtrl[idx].lastTime = g_actTime;
                   runCtrl[idx].actEvt++;
               }
               else
                   factPrintf(kError, 503, "writeEvt: Writing event for run %d failed (runWrite)", irun);

               checkAndCloseRun(idx, rc<0, 1);
           }

           evtCtrl[k0].evtStat = 10000;  // event written (or has to be discarded) -> delete
       }

       // Although the are no pending events, we have to check if a run should be closed (timeout)
       for (int j=0; j<MAX_RUN; j++)
       {
           if (runCtrl[j].fileId == 0)
           {
               //ETIENNE added the condition at this line. dunno what to do with run 0: skipping it
               const int cond = /*runCtrl[j].lastTime < lastStartedTime &&*/ runCtrl[j].runId == 0;
               checkAndCloseRun(j, cond, 2);
           }
       }

       usleep(1);

       //nothing left to do
       if (gj.readStat < -10 && numWait == 0)
       {
           factPrintf(kInfo, -1, "Finish Write Process ...");
           gj.writStat = -22;     //==> we should exit
           break;
       }

       gj.writStat = gj.readStat;
   }

   factPrintf(kInfo, -1, "Close all open files ...");
   for (int j=0; j<MAX_RUN; j++)
   {
       if (runCtrl[j].fileId == 0)
           checkAndCloseRun(j, 1, 3);
   }

   gj.writStat = -99;

   factPrintf(kInfo, -1, "Exit Writing Process ...");

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




void
StartEvtBuild ()
{

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

   gj.readStat = gj.procStat = gj.writStat = 0;

   factPrintf(kInfo, -1, "Starting EventBuilder V15.07 A");

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

   gi_maxProc = g_maxProc;
   if (gi_maxProc <= 0 || gi_maxProc > 90) {
      factPrintf(kFatal, 301, "Illegal number of processes %d", gi_maxProc);
      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,
            int8_t * 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 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 * 200;   //100MBytes

   g_maxProc = 20;

   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