// // // #define EVTDEBUG #define NUMSOCK 1 //set to 7 for old configuration #define MAXREAD 65536 //64kB wiznet buffer #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #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 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 g_maxProc; int g_maxSize; int gi_maxSize; int gi_maxProc; 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 #define MXSTR 1000 char str[MXSTR]; //ETIENNE #define MAX_SLOTS_PER_CHUNK 100 typedef struct { int32_t eventNumber; int32_t chunk; int32_t slot; } CHUNK_MAPPING; CHUNK_MAPPING mBufferMapping[MAX_EVT * MAX_RUN]; #define MAX_EVT_MEM (sizeof(EVENT) + NPIX*1024*2 + NTMARK*1024*2) #define MAX_HEAD_MEM (NBOARDS * sizeof(PEVNT_HEADER)) #define MAX_SLOT_SIZE (MAX_EVT_MEM + MAX_HEAD_MEM) #define MAX_CHUNK_SIZE (MAX_SLOT_SIZE*MAX_SLOTS_PER_CHUNK) typedef struct { void * pointers[MAX_SLOTS_PER_CHUNK]; int32_t events[MAX_SLOTS_PER_CHUNK]; int32_t nFreeSlots; int32_t nSlots; } ETI_CHUNK; int32_t numAllocatedChunks = 0; #define MAX_CHUNKS 8096 ETI_CHUNK EtiMemoryChunks[MAX_CHUNKS]; void* ETI_Malloc(int evtId, int evtIndex) { for (int i=0;i 0) { for (int j=0;jevents[mBufferMapping[evtIndex].slot] != evtId) { snprintf (str, MXSTR, "Mismatch in chunk mapping table. Expected evtId %d. Got %d. No memory was freed.", evtId, currentChunk->events[mBufferMapping[evtIndex].slot]); factOut (kError, 000, str); return; } currentChunk->events[mBufferMapping[evtIndex].slot] = -1; currentChunk->nFreeSlots++; int chunkIndex = mBufferMapping[evtIndex].chunk; if (chunkIndex != numAllocatedChunks-1) return; while (EtiMemoryChunks[chunkIndex].nFreeSlots == EtiMemoryChunks[chunkIndex].nSlots) {//free this chunk if (EtiMemoryChunks[chunkIndex].pointers[0] == NULL) { snprintf(str, MXSTR, "Chunk %d not allocated as it ought to be. Skipping memory release.", chunkIndex); factOut(kError, 000, str); return; } free(EtiMemoryChunks[chunkIndex].pointers[0]); EtiMemoryChunks[chunkIndex].pointers[0] = NULL; numAllocatedChunks--; chunkIndex--; if (numAllocatedChunks == 0) break; } } //END ETIENNE 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 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 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; 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 runFinish (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 (sid % 7 >= NUMSOCK) { //this is a not used socket, so do nothing ... rd->sockStat = 77; rd->rBuf = NULL ; return 0; } if (rd->sockStat == 0) { //close socket if open j = close (rd->socket); if (j > 0) { snprintf (str, MXSTR, "Closing socket %d failed: %m (close,rc=%d)", sid, errno); factOut (kFatal, 771, str); } else { snprintf (str, MXSTR, "Succesfully closed socket %d", sid); factOut (kInfo, 771, str); } } rd->sockStat = 99; if (flag < 0) { free (rd->rBuf); //and never open again #ifdef EVTDEBUG free (rd->xBuf); //and never open again #endif rd->rBuf = NULL; 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 (malloc failed)", 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, "Generating socket %d failed: %m (socket,rc=%d)", sid, errno); 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, "Setting SO_KEEPALIVE for socket %d failed: %m (setsockopt,rc=%d)", sid, errno); 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, "Setting TCP_KEEPIDLE for socket %d failed: %m (setsockopt,rc=%d)", sid, errno); 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, "Setting TCP_KEEPINTVL for socket %d failed: %m (setsockopt,rc=%d)", sid, errno); 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, "Setting TCP_KEEPCNT for socket %d failed: %m (setsockopt,rc=%d)", sid, errno); 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 rd->errCnt = 0; return 0; } /*-----------------------------------------------------------------*/ /*-----------------------------------------------------------------*/ int mBufInit () { // initialize mBuffer (mark all entries as unused\empty) uint32_t actime = g_actTime + 50000000; for (int 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 //ETIENNE mBufferMapping[i].chunk = -1; mBufferMapping[i].eventNumber = -1; mBufferMapping[i].slot = -1; //END ETIENNE } for (int j=0;j 1024) { snprintf (str, MXSTR, "Illegal roi in channel 0: %d (allowed: 0<=roi<=1024)", nRoi[0]); factOut (kError, 999, str); gj.badRoiR++; gj.badRoi[b]++; return -9999; } for (int jr = 1; jr < 8; jr++) { if (nRoi[jr] != nRoi[0]) { snprintf (str, MXSTR, "Mismatch of roi (%d) in channel %d with roi (%d) in channel 0.", nRoi[jr], jr, nRoi[0]); factOut (kError, 711, str); gj.badRoiB++; gj.badRoi[b]++; return -7101; } } if (nRoi[8] < nRoi[0]) { snprintf (str, MXSTR, "Mismatch of roi (%d) in channel 8. Should be larger or equal than the roi (%d) in channel 0.", nRoi[8], nRoi[0]); factOut (kError, 712, str); gj.badRoiB++; gj.badRoi[b]++; return -7102; } i = evID % MAX_EVT; evFree = -1; for (int 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, "Mismatch of roi within event. Expected roi=%d and roi_tm=%d, got %d and %d.", mBuffer[i].nRoi, mBuffer[i].nRoiTM, nRoi[0], nRoi[8]); 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 (int 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 finished", runID); // factOut (kInfo, 931, str); // return -21; // } if (runCtrl[k].roi0 != nRoi[0] || runCtrl[k].roi8 != nRoi[8]) { snprintf (str, MXSTR, "Mismatch of roi within run. Expected roi=%d and roi_tm=%d, got %d and %d.", runCtrl[k].roi0, runCtrl[k].roi8, nRoi[0], nRoi[8]); 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, "New run %d (evFree=%d) registered with roi=%d and roi_tm=%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 // runCtrl[evFree].lastTime = 0; runTail[evFree].nEventsOk = runTail[evFree].nEventsRej = runTail[evFree].nEventsBad = runTail[evFree].PCtime0 = runTail[evFree].PCtimeX = 0; } RUNFOUND: //ETIENNE /* 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 = ETI_Malloc(evID, i); mBuffer[i].buffer = NULL; if (mBuffer[i].FADhead != NULL) mBuffer[i].fEvent = (EVENT*)&(((char*)(mBuffer[i].FADhead))[MAX_HEAD_MEM]); else { mBuffer[i].fEvent = NULL; gj.usdMem = 0; for (int k=0;k gj.maxMem) gj.maxMem = gj.usdMem; 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; } 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; }*/ //END ETIENNE //flag all boards as unused mBuffer[i].nBoard = 0; for (int k = 0; k < NBOARDS; k++) { mBuffer[i].board[k] = -1; } //flag all pixels as unused for (int k = 0; k < NPIX; k++) { mBuffer[i].fEvent->StartPix[k] = -1; } //flag all TMark as unused for (int 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; //ETIENNE //gj.usdMem += needmem + headmem + gi_maxSize; gj.usdMem = 0; for (int k=0;k 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; //ETIENNE ETI_Free(evid, i); // free (mBuffer[i].fEvent); mBuffer[i].fEvent = NULL; // free (mBuffer[i].FADhead); mBuffer[i].FADhead = NULL; // free (mBuffer[i].buffer); mBuffer[i].buffer = NULL; //END ETIENNE // headmem = NBOARDS * sizeof (PEVNT_HEADER); mBuffer[i].evNum = mBuffer[i].nRoi = -1; mBuffer[i].runNum = 0; //ETIENNE // gj.usdMem = gj.usdMem - freemem - headmem - gi_maxSize; gj.usdMem = 0; for (int k=0;k 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; 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; for (b = 0; b < NBOARDS; b++) gj.badRoi[b] = 0; mBufInit (); //initialize buffers snprintf (str, MXSTR, "End initializing (readFAD)"); 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++) { 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 if (rd[i].sockStat == -1) { rd[i].sockStat = connect (rd[i].socket, (struct sockaddr *) &rd[i].SockAddr, sizeof (rd[i].SockAddr)); if (rd[i].sockStat == -1) { rd[i].errCnt++ ; // if (rd[i].errCnt < 100) rd[i].sockStat = rd[i].errCnt ; // else if (rd[i].errCnt < 1000) rd[i].sockStat = 1000 ; // else rd[i].sockStat = 10000 ; } //printf("try to connect %d -> %d\n",i,rd[i].sockStat); } if (rd[i].sockStat < -1 ) { rd[i].sockStat++ ; } 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]++; gi_NumConnect[b] += cntsock ; gi.numConn[b]++; gj.numConn[b]++; snprintf (str, MXSTR, "New connection %d (number of connections: %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); // rd[i].bufLen, 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]--; 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, "Reading from socket %d failed: %m (recv,rc=%d)", i, errno); 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, "End-of-event flag wrong on socket %3d for event %4d (len=%5d), expected %3d %3d, got %3d %3d", i, rd[i].evtID, rd[i].fadLen, stop.B[0], stop.B[1], rd[i].rBuf->B[rd[i].bufPos - 1], rd[i].rBuf->B[rd[i].bufPos - 2]); 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, kr; int checkRoi; roi[0] = ntohs (rd[i].rBuf->S[head_len / 2 + 2]); for (kr = 1; kr < 4; kr++) { checkRoi = ntohs(rd[i].rBuf->S[head_len/ 2 + + kr*(roi[jr-1]+4)]); if (checkRoi != roi[0]) { snprintf (str, MXSTR, "Inconsistent Roi accross board patches"); factOut (kFatal, 1, str); goto EndBuf; } } for (jr = 1; jr < 9; jr++) { roi[jr] = ntohs (rd[i]. rBuf->S[head_len / 2 + 2 + jr * (roi[jr-1] + 4)]); } */ //we have a complete buffer, copy to WORK area //End of the header. to channels now int eStart = 36; for (int ePatchesCount = 0; ePatchesCount<4*9;ePatchesCount++) { rd[i].rBuf->S[eStart+0] = ntohs(rd[i].rBuf->S[eStart+0]);//id rd[i].rBuf->S[eStart+1] = ntohs(rd[i].rBuf->S[eStart+1]);//start_cell rd[i].rBuf->S[eStart+2] = ntohs(rd[i].rBuf->S[eStart+2]);//roi rd[i].rBuf->S[eStart+3] = ntohs(rd[i].rBuf->S[eStart+3]);//filling eStart += 4+rd[i].rBuf->S[eStart+2];//skip the pixel data } //channels done. footer now // rd[i].rBuf->S[eStart+eCount] = ntohs(rd[i].rBuf->S[eStart+eCount]);//package_crc // eCount++; // rd[i].rBuf->S[eStart+eCount] = ntohs(rd[i].rBuf->S[eStart+eCount]);//end_package_flag // snprintf(str, MXSTR, "Bytes read %d bytes swapped %d", jrd, eCount*2); // factOut(kInfo, 000, str); //ETIENNE end of bytes swapping already int jr, kr; int checkRoi; int roiHopper = head_len/2 + 2; //points to the very first roi roi[0] = rd[i].rBuf->S[roiHopper]; roiHopper += roi[0]+4;//skip to the second roi (i.e. next board, same patch-pixel) for (kr = 1; kr < 4; kr++) { checkRoi = rd[i].rBuf->S[roiHopper]; if (checkRoi != roi[0]) { snprintf (str, MXSTR, "Inconsistent Roi accross boards B=%d, expected %d, got %d", kr, checkRoi, roi[0]); factOut (kError, 1, str); goto EndBuf; } roiHopper += checkRoi+4; } //roiHopper now points to the first pixel of board 2. Do the 8 remaining pixels for (jr = 1; jr < 9; jr++) { roi[jr] = rd[i].rBuf->S[roiHopper]; checkRoi = roi[jr]; for (kr = 1; kr < 4; kr++) { roiHopper += checkRoi+4; checkRoi = rd[i].rBuf->S[roiHopper]; if (checkRoi != roi[jr]) { snprintf (str, MXSTR, "Inconsistent Roi accross patches B=%d P=%d, expected %d, got %d", kr, jr, roi[jr], checkRoi); factOut (kFatal, 1, str); goto EndBuf; } } } //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 = rd[i].rBuf->S[12]; int fadCrate = fadBoard / 256; if (boardId != (fadCrate * 10 + fadBoard % 256)) { snprintf (str, MXSTR, "Board ID mismatch. Expected %d, got %d (C=%d, B=%d)", boardId, fadBoard, fadCrate, fadBoard % 256); factOut (kWarn, 301, str); } if (mBuffer[evID].board[boardId] != -1) { snprintf (str, MXSTR, "Got event %5d from board %3d (i=%3d, len=%5d) twice: Starts with %3d %3d - ends with %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 = rd[i].rBuf->S[src++]; // ID src++; pixC = rd[i].rBuf->S[src++]; // start-cell pixR = 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; //ETIENNE because the TM channels are always processed during drs calib, //set them to zero if they are not present //I suspect that it may be more efficient to set all the allocated mem to //zero when allocating it // dest = tmS*roi[0] + NPIX*roi[0]; // bzero(&mBuffer[evID].fEvent->Adc_Data[dest],roi[0]*2); } } } } // 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 called for run %d, event %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++; // 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]; rd[i].bufTyp = 1; //ready to read full record rd[i].bufLen = rd[i].fadLen - rd[i].bufPos; #ifdef EVTDEBUG int fadboard = rd[i].rBuf->S[12]; int fadcrate = fadboard / 256; fadboard = (fadcrate * 10 + fadboard % 256); 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 %d on port %d, %d expected.", rd[i].bufLen, i, head_len); factOut (kFatal, 881, str); rd[i].bufLen = 100000; //? } int fadBoard = 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 - 30) { int id = evtCtrl.evtBuf[k0]; snprintf (str, MXSTR, "%5d skip incomplete evt %8d %8d %2d", mBuffer[id].evNum, evtCtrl.evtBuf[k0], k0, evtCtrl.evtStat[k0]); factOut (kWarn, 601, str); int ik,ib,jb; ik=0; for (ib=0; ib0 ) { snprintf (&str[ik], MXSTR, "."); } else { snprintf (&str[ik], MXSTR, "x"); } } else { snprintf (&str[ik], MXSTR, "%d",jb%10); } ik++; } snprintf (&str[ik], MXSTR, "|"); factOut (kWarn, 601, str); evtCtrl.evtStat[k0] = 91; //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]--; 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) { int64_t threadID; int numWait, numProc, k, jret; struct timespec xwait; // int32_t cntr ; threadID = (int64_t)thrid; snprintf (str, MXSTR, "Starting sub-process-thread %ld", 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].buffer); if (jret <= threadID) { snprintf (str, MXSTR, "Process %ld wants to send event to process %d... not allowed.", 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 in process %ld", 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 %ld", threadID); factOut (kInfo, -1, str); return 0; } /*-----------------------------------------------------------------*/ void * procEvt (void *ptr) { /* *** main loop processing file, including SW-trigger */ int numProc, numWait; int 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 subprocesses", 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 (long long 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 (int 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 { //-------- 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) int id = evtCtrl.evtBuf[k0]; uint32_t irun = mBuffer[id].runNum; int32_t ievt = mBuffer[id].evNum; if (runCtrl[lastRun].runId == irun) { j = lastRun; } else { //check which fileID to use (or open if needed) for (j = 0; j < MAX_RUN; j++) { if (runCtrl[j].runId == irun) break; } if (j >= MAX_RUN) { snprintf (str, MXSTR, "procEvt: Can not find run %d for event %d in %d", irun, ievt, id); factOut (kFatal, 901, str); } lastRun = j; } if (runCtrl[j].fileId < 0) { //---- we need to open a new run ==> make sure all older runs are //---- finished and marked to be closed .... int j1; for (j1 = 0; j1 < MAX_RUN; j1++) { if (runCtrl[j1].fileId == 0) { runCtrl[j1].procId = 2; //--> do no longer accept events for processing //---- problem: processing still going on ==> must wait for closing .... snprintf (str, MXSTR, "procEvt: Finished run since new one opened %d", runCtrl[j1].runId); runFinish1 (runCtrl[j1].runId); } } actRun.Version = 1; actRun.RunType = -1; //to be adapted actRun.Nroi = runCtrl[j].roi0; actRun.NroiTM = runCtrl[j].roi8; //ETIENNE don't reset it to zero as it is taken care of in DataWriteFits // 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, "procEvt: Could not open a file for run %d (runOpen failed)", irun); factOut (kError, 502, str); runCtrl[j].fileId = 91; runCtrl[j].procId = 91; } else { snprintf (str, MXSTR, "procEvt: Opened new file for run %d (evt=%d)", irun, ievt); factOut (kInfo, -1, str); runCtrl[j].fileId = 0; runCtrl[j].procId = 0; } } //-------- also check if run shall be closed (==> skip event, but do not close the file !!! ) if (runCtrl[j].procId == 0) { if (runCtrl[j].closeTime < g_actTime || runCtrl[j].lastTime < g_actTime - 300 || runCtrl[j].maxEvt <= runCtrl[j].procEvt) { snprintf (str, MXSTR, "procEvt: Reached end of run condition for run %d", irun); factOut (kInfo, 502, str); runFinish1 (runCtrl[j].runId); runCtrl[j].procId = 1; } } if (runCtrl[j].procId != 0) { snprintf (str, MXSTR, "procEvt: Skip event %d because no active run %d", ievt, irun); factOut (kDebug, 502, str); evtCtrl.evtStat[k0] = 9091; } else { //-------- //-------- 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 //ETIENNE don't reset it to zero as it is taken care of in DataWriteFits // 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 ; also check for consistency in event (not yet) 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; 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] = mBuffer[id].FADhead[ib].time; } } int i = eventCheck (mBuffer[id].runNum, mBuffer[id].FADhead, mBuffer[id].fEvent); gi.procTot++; numProc++; if (i < 0) { evtCtrl.evtStat[k0] = 9999; //flag event to be skipped gi.procErr++; } else { evtCtrl.evtStat[k0] = 1000; runCtrl[j].procEvt++; } } } } 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 (int k = 0; k < gi_maxProc; k++) { pthread_join (thread[k], (void **) &status); } int k1 = evtCtrl.frstPtr; for (int 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 ; 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] < 9000) { 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, "writeEvt: 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; //ETIENNE don't reset it to zero as it is taken care of in DataWriteFits // 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, "writeEvt: Could not open a file for run %d (runOpen failed)", irun); factOut (kError, 502, str); runCtrl[j].fileId = 91; } else { snprintf (str, MXSTR, "writeEvt: Opened new file for run %d (evt %d)", irun, ievt); factOut (kInfo, -1, str); runCtrl[j].fileId = 0; } } if (runCtrl[j].fileId != 0) { if (runCtrl[j].fileId < 0) { snprintf (str, MXSTR, "writeEvt: Never opened file for run %d", irun); factOut (kError, 123, str); } else if (runCtrl[j].fileId < 100) { snprintf (str, MXSTR, "writeEvt: File for run %d is closed", irun); factOut (kWarn, 123, str); runCtrl[j].fileId += 100; } else { snprintf (str, MXSTR, "writeEvt: File for run %d is closed", irun); factOut (kDebug, 123, str); } evtCtrl.evtStat[k0] = 9903; gi.wrtErr++; } else { // snprintf (str, MXSTR,"write event %d size %d",ievt,sizeof (mBuffer[id])); // factOut (kInfo, 504, str); int 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); // gj.writEvt++ ; } else { snprintf (str, MXSTR, "writeEvt: Writing event for run %d failed (runWrite)", 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, "writeEvt: Error closing run %d (runClose[1],i=%d)", runCtrl[j].runId, i); factOut (kError, 503, str); runCtrl[j].fileId = 92; } else { snprintf (str, MXSTR, "writeEvt: Closed run %d (ii[1]=%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) //ETIENNE but first figure out which one is the latest run with a complete event. //i.e. max run Id and lastEvt >= 0 //this condition is sufficient because all pending events were written already in the loop just above unsigned int maxStartedRun = 0; for (j=0;j maxStartedRun) && (runCtrl[j].lastEvt > -1) && (runCtrl[j].runId != 0)) maxStartedRun = runCtrl[j].runId; } // snprintf(str, MXSTR, "Maximum runId: %d", maxStartedRun); // factOut(kInfo, 000, str); //Also check if some files will never be opened //EDIT: this is completely useless, because as run Numbers are taken from FADs board, //I will never get run numbers for which no file is to be opened for (j=0;j 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; int 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, "writeEvt: Error closing run %d (runClose[3],i=%d)", runCtrl[j].runId, i); factOut (kError, 506, str); runCtrl[j].fileId = 96; } else { snprintf (str, MXSTR, "writeEvt: Closed run %d (ii[3]=%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 = 1; 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, 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