Index: trunk/FACT++/src/EventBuilder.c =================================================================== --- trunk/FACT++/src/EventBuilder.c (revision 15356) +++ trunk/FACT++/src/EventBuilder.c (revision 15360) @@ -102,5 +102,4 @@ //int gi_memStat = +1; -uint32_t gi_myRun = 0; uint32_t actrun = 0; @@ -366,8 +365,4 @@ CNV_FACT *rBuf; -#ifdef EVTDEBUG - CNV_FACT *xBuf; //a copy of rBuf (temporary for debuging) -#endif - } READ_STRUCT; @@ -448,7 +443,4 @@ 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; @@ -462,7 +454,4 @@ 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) { @@ -546,10 +535,9 @@ } /*-----------------------------------------------------------------*/ - - - -int -mBufEvt (int evID, uint runID, int nRoi[], int sk, - int fadlen, int trgTyp, int trgNum, int fadNum) +void swapEventHeaderBytes(int i); +int checkRoiConsistency(int i, int roi[]); + +int mBufEvt (int sk) +/*(int evID, uint runID, int nRoi[], int sk, int fadlen, int trgTyp, int trgNum, int fadNum)*/ { // generate a new Event into mBuffer: @@ -565,4 +553,9 @@ // size_t needmem = 0; + swapEventHeaderBytes(sk); + + int nRoi[9]; + if (!checkRoiConsistency(sk, nRoi)) + return -9999; const int b = sk / 7; @@ -590,4 +583,10 @@ } + const int evID = rd[sk].evtID; + const uint runID = rd[sk].runID; + const int fadlen = rd[sk].fadLen; + const int trgTyp = rd[sk].ftmTyp; + const int trgNum = rd[sk].ftmID; + const int fadNum = rd[sk].evtID; int i = evID % MAX_EVT; @@ -1020,4 +1019,126 @@ } +int checkRoiConsistency(int i, int roi[]) +{ + int xjr = -1; + int xkr = -1; + + //points to the very first roi + int roiPtr = sizeof(PEVNT_HEADER)/2 + 2; + + roi[0] = rd[i].rBuf->S[roiPtr]; + + for (int jr = 0; jr < 9; jr++) + { + roi[jr] = rd[i].rBuf->S[roiPtr]; + + if (roi[jr]!=roi[0]) + { + xjr = jr; + break; + } + + for (int kr = 1; kr < 4; kr++) + { + const int kroi = rd[i].rBuf->S[roiPtr]; + if (kroi != roi[jr]) + { + xjr = jr; + xkr = kr; + break; + } + roiPtr += kroi+4; + } + } + + if (xjr<0) + return 1; + + if (xkr<0) + factPrintf(kFatal, 1, "Inconsistent Roi accross boards B=%d, expected %d, got %d", xjr, roi[xjr], roi[0]); + else + factPrintf(kFatal, 1, "Inconsistent Roi accross patches B=%d P=%d, expected %d, got %d", xjr, xkr, roi[xjr], rd[i].rBuf->S[roiPtr]); + + return 0; +} + +void swapEventHeaderBytes(int i) +{ + //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 + } +} + +void copyData(int i, int evID, /*int roi,*/ int boardId) +{ + memcpy(&mBuffer[evID].FADhead[boardId].start_package_flag, + &rd[i].rBuf->S[0], sizeof(PEVNT_HEADER)); + + int src = sizeof(PEVNT_HEADER) / 2; + + // consistency of ROIs have been checked already (is it all correct?) + const int roi = rd[i].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 = rd[i].rBuf->S[src++]; // start-cell + const int pixR = rd[i].rBuf->S[src++]; // roi + //here we should check if pixH is correct .... + + const int pixS = boardId * 36 + drs * 9 + px; + src++; + + mBuffer[evID].fEvent->StartPix[pixS] = pixC; + + const int dest1 = pixS * roi; + memcpy(&mBuffer[evID].fEvent->Adc_Data[dest1], + &rd[i].rBuf->S[src], roi * 2); + + src += pixR; + + if (px == 8) + { + const int tmS = boardId * 4 + drs; + + //and we have additional TM info + if (pixR > roi) + { + const int dest2 = tmS * roi + NPIX * roi; + + const int srcT = src - roi; + mBuffer[evID].fEvent->StartTM[tmS] = (pixC + pixR - roi) % 1024; + + memcpy(&mBuffer[evID].fEvent->Adc_Data[dest2], + &rd[i].rBuf->S[srcT], roi * 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); + } + } + } + } +} @@ -1044,18 +1165,9 @@ -void * -readFAD (void *ptr) +void *readFAD (void *ptr) { /* *** main loop reading FAD data and sorting them to complete events */ - int actBoards = 0; - uint gi_SecTime; //time in seconds - - int sockDef[NBOARDS]; //internal state of sockets - - struct timeval tv; - uint32_t tsec, tusec; - - - factPrintf(kInfo, -1, "Start initializing (readFAD)"); + + factPrintf(kInfo, -1, "Start initializing (readFAD)"); // int cpu = 7; //read thread @@ -1075,6 +1187,7 @@ + const int minLen = sizeof(PEVNT_HEADER); //min #bytes needed to check header: full header for debug + int frst_len = sizeof(PEVNT_HEADER); //max #bytes to read first: fad_header only, so each event must be longer, even for roi=0 - int minLen = sizeof(PEVNT_HEADER); //min #bytes needed to check header: full header for debug start.S = 0xFB01; @@ -1088,18 +1201,14 @@ } gi_resetS = gi_resetR = 9; - for (int i = 0; i < NBOARDS; i++) - sockDef[i] = 0; + + int sockDef[NBOARDS]; //internal state of sockets + memset(sockDef, 0, NBOARDS*sizeof(int)); START: - gettimeofday (&tv, NULL); - g_actTime = tsec = tv.tv_sec; - g_actUsec = tusec = tv.tv_usec; - gi_myRun = g_actTime; evtCtrl.frstPtr = 0; evtCtrl.lastPtr = 0; - gi_SecTime = g_actTime; - gi_runStat = g_runStat; - gj.readStat = g_runStat; + //time in seconds + uint gi_SecTime = time(NULL);; const int cntsock = 8 - NUMSOCK ; @@ -1148,55 +1257,51 @@ gi_reset = gi_resetR = gi_resetS = gi_resetW = 0; - while (g_runStat >= 0 && g_reset == 0) { //loop until global variable g_runStat claims stop - + //loop until global variable g_runStat claims stop + while (g_runStat >= 0 && g_reset == 0) + { gi_runStat = g_runStat; gj.readStat = g_runStat; + + struct timeval tv; gettimeofday (&tv, NULL); - g_actTime = tsec = tv.tv_sec; - g_actUsec = tusec = tv.tv_usec; - - - int nch = 0; - for (int b = 0; b < NBOARDS; b++) { - int 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; - - int s0; - 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 - - const int p0 = s0==0 ? ntohs (g_port[b].sockAddr.sin_port) : 0; - - for (int 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; - } + g_actTime = tv.tv_sec; + g_actUsec = tv.tv_usec; + + + 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 + gi.numConn[b] = 0; + 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; + + int k = b * 7; + for (int p = p0 + 1; p < p0 + 8; p++, k++) + GenSock (s0, k, p, &g_port[b].sockAddr, &rd[k]); //generate address and socket + + sockDef[b] = g_port[b].sockDef; } - if (nch > 0) { - actBoards = 0; - for (int b = 0; b < NBOARDS; b++) { - if (sockDef[b] > 0) - actBoards++; - } - } - - -#ifdef EVTDEBUG - int jrdx = 0; -#endif - - int numokx = 0; - int numok = 0; //count number of succesfull actions + // count the number of active boards + int actBoards = 0; + for (int b = 0; b < NBOARDS; b++) + if (sockDef[b] > 0) + actBoards++; + + //count number of succesfull actions + int numok = 0; /* @@ -1216,558 +1321,306 @@ */ - for (int i = 0; i < MAX_SOCK; i+=7) { //check all sockets if something to read + //check all sockets if something to read + for (int i = 0; i < MAX_SOCK; i+=7) + { + // Do not try to connect this socket + if (rd[i].sockStat > 0) + continue; const int b = i / 7 ; //const int p = i % 7 ; -/*if ( p >= NUMSOCK) { ; } - else*/ { - const int s0 = sockDef[b] > 0 ? +1 : -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++ ; - usleep(25000); -// 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]++; - factPrintf(kInfo, -1, "New connection %d (number of connections: %d)", b, gi.numConn[b]); - } - } - - if (rd[i].sockStat == 0) { //we have a connection ==> try to read - - if (rd[i].bufLen<=0) - { + 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[b] > 0) + { + rd[i].bufTyp = 0; // expect a header + rd[i].bufLen = frst_len; // max size to read at begining + } + else + { + rd[i].bufTyp = -1; // full data to be skipped + rd[i].bufLen = MAX_LEN; // huge for skipping + } + + rd[i].bufPos = 0; // no byte read so far + rd[i].skip = 0; // start empty + + gi_NumConnect[b] += cntsock; + + gi.numConn[b]++; + gj.numConn[b]++; + + factPrintf(kInfo, -1, "New connection %d (number of connections: %d)", b, gi.numConn[b]); + } + + // Do not read from this socket + if (rd[i].bufLen<=0) + continue; + + numok++; + + 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) + { + numok--; + continue; + } + + factPrintf(kError, 442, "Reading from socket %d failed: %m (recv,rc=%d)", i, errno); + gi.gotErr[b]++; + continue; + } + + // connection was closed ... + if (jrd==0) + { + factPrintf(kInfo, 441, "Socket %d closed by FAD", i); + + const int s0 = sockDef[b] > 0 ? +1 : -1; + GenSock(s0, i, 0, NULL, &rd[i]); + + gi.gotErr[b]++; + + gi_NumConnect[b]-= cntsock ; + gi.numConn[b]--; + gj.numConn[b]--; + + continue; + } + + // Success (jrd > 0) + + gi.gotByte[b] += jrd; + gj.rateBytes[b] += jrd; + + // are we skipping this board ... + if (rd[i].bufTyp < 0) + continue; + + rd[i].bufPos += jrd; //==> prepare for continuation + rd[i].bufLen -= jrd; + + // are we reading data? + if (rd[i].bufTyp > 0) + { + // not yet all read + if (rd[i].bufLen > 0) + continue; + + // ready to read next header + rd[i].bufTyp = 0; + rd[i].bufLen = frst_len; + rd[i].bufPos = 0; + + if (rd[i].rBuf->B[rd[i].fadLen - 1] != stop.B[0] || + rd[i].rBuf->B[rd[i].fadLen - 2] != stop.B[1]) + { + gi.evtErr++; + factPrintf(kError, 301, "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].fadLen - 1], rd[i].rBuf->B[rd[i].fadLen - 2]); + continue; + } + #ifdef EVTDEBUG - factPrintf(kDebug, 301, "do not read from socket %d %d", i, rd[i].bufLen); + debugRead (i, jrd, rd[i].evtID, rd[i].ftmID, rd[i].runID, 1, tv.tv_sec, tv.tv_usec); // i=socket; jrd=#bytes; ievt=eventid; 1=finished event #endif - continue; - } - -// if (rd[i].bufLen > 0) { //might be nothing to read [buffer full] - numok++; - - const int32_t jrd = - recv (rd[i].socket, &rd[i].rBuf->B[rd[i].bufPos], - rd[i].bufLen, MSG_DONTWAIT); - -#ifdef EVTDEBUG - if (jrd > 0) { - debugStream (i, &rd[i].rBuf->B[rd[i].bufPos], jrd); - memcpy (&rd[i].xBuf->B[rd[i].bufPos], - &rd[i].rBuf->B[rd[i].bufPos], jrd); - factPrintf(kDebug, 301, "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]); - } -#endif - - if (jrd == 0) { //connection has closed ... - factPrintf(kInfo, 441, "Socket %d closed by FAD", i); - 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) { - factPrintf(kError, 442, "Reading from socket %d failed: %m (recv,rc=%d)", i, errno); - gi.gotErr[b]++; - } else - numok--; //else nothing waiting to be read - //jrd = 0; - } -/* } else { - // jrd = 0; //did read nothing as requested -#ifdef EVTDEBUG - snprintf (str, MXSTR, "do not read from socket %d %d", i, - rd[i].bufLen); - factOut (kDebug, 301, str); -#endif - } -*/ - if (jrd<=0) - continue; - - gi.gotByte[b] += jrd; - gj.rateBytes[b] += jrd; - - //if (jrd > 0) { - numokx++; -#ifdef EVTDEBUG - jrdx += jrd; -#endif - //} - - - if (rd[i].bufTyp < 0) { // we are skipping this board ... -// just do nothing -#ifdef EVTDEBUG - factPrintf(kInfo, 301, "skipping %d bytes on socket %d", jrd, i); -#endif - continue; - } - - rd[i].bufPos += jrd; //==> prepare for continuation - rd[i].bufLen -= jrd; - - if (rd[i].bufTyp > 0) { // we are reading data ... - - if (rd[i].bufLen>0/*jrd < rd[i].bufLen*/) { //not yet all read -#ifdef EVTDEBUG - debugRead (i, jrd, rd[i].evtID, rd[i].ftmID, rd[i].runID, 0, tsec, tusec); // i=socket; jrd=#bytes; ievt=eventid; 0=reading data -#endif + + // 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 evID = mBufEvt(i); + + if (evID < -1000) continue; - } - - //int pos = rdi_bufPos = rd[i].bufPos; - //int pos = rdi_bufLen = rd[i].bufLen; - //int pos = rdi_bufTyp = rd[i].bufTyp; - - rd[i].bufTyp = 0; //ready to read next header - rd[i].bufLen = frst_len; - rd[i].bufPos = 0; - - //{ - //full dataset read - //rd[i].bufLen = 0; - //rd[i].bufPos = rd[i].fadLen; - if (rd[i].rBuf->B[rd[i].fadLen - 1] != stop.B[0] || - rd[i].rBuf->B[rd[i].fadLen - 2] != stop.B[1]) { - gi.evtErr++; - factPrintf(kError, 301, "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].fadLen - 1], rd[i].rBuf->B[rd[i].fadLen - 2]); - //goto EndBuf; - continue; - -#ifdef EVTDEBUG - } else { - factPrintf(kDebug, 301, "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].fadLen - 2], - rd[i].rBuf->B[rd[i].fadLen - 1], stop.B[1], - stop.B[0]); -#endif - } - -#ifdef EVTDEBUG - //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 -#endif - -/* //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++) + + // no space left, retry later + if (evID < 0) + { + rd[i].bufTyp = -1; + rd[i].bufLen = 0; + rd[i].bufPos = rd[i].fadLen; + continue; + } + + //we have a valid entry in mBuffer[]; fill it + + const int boardId = b; + const int fadBoard = rd[i].rBuf->S[12]; + const int fadCrate = fadBoard / 256; + + if (boardId != (fadCrate * 10 + fadBoard % 256)) + { + factPrintf(kWarn, 301, "Board ID mismatch. Expected %d, got %d (C=%d, B=%d)", + boardId, fadBoard, fadCrate, fadBoard % 256); + } + + if (mBuffer[evID].board[boardId] != -1) + { + factPrintf(kWarn, 501, "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].fadLen - 2], + rd[i].rBuf->B[rd[i].fadLen - 1]); + // FIXME: Is that action the right one? + continue; + } + + // Copy data from rd[i] to mBuffer[evID] + copyData(i, evID, boardId); + + // now we have stored a new board contents into Event structure + + const int iDx = evtIdx[evID]; //index into evtCtrl + + evtCtrl.evtStat[iDx]++; + evtCtrl.pcTime[iDx] = g_actTime; + + mBuffer[evID].fEvent->NumBoards++; + mBuffer[evID].board[boardId] = boardId; + mBuffer[evID].nBoard++; + + // have we already reported first (partial) event of this run ??? + if (mBuffer[evID].nBoard>0 && mBuffer[evID].runNum != actrun) + { + actrun = mBuffer[evID].runNum; + + for (int ir = 0; ir < MAX_RUN; ir++) { - checkRoi = ntohs(rd[i].rBuf->S[head_len/ 2 + - + kr*(roi[jr-1]+4)]); - if (checkRoi != roi[0]) + if (runCtrl[ir].runId == actrun) { - 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 checkRoi; - int roiHopper = sizeof(PEVNT_HEADER)/2 + 2; //points to the very first roi - int roi[9]; - roi[0] = rd[i].rBuf->S[roiHopper]; - roiHopper += roi[0]+4;//skip to the second roi (i.e. next board, same patch-pixel) - for (int kr = 1; kr < 4; kr++) - { - checkRoi = rd[i].rBuf->S[roiHopper]; - if (checkRoi != roi[0]) - { - factPrintf(kError, 1, "Inconsistent Roi accross boards B=%d, expected %d, got %d", kr, checkRoi, roi[0]); -// goto EndBuf; - continue; - } - roiHopper += checkRoi+4; - } - //roiHopper now points to the first pixel of board 2. Do the 8 remaining pixels - for (int jr = 1; jr < 9; jr++) { - roi[jr] = rd[i].rBuf->S[roiHopper]; - checkRoi = roi[jr]; - for (int kr = 1; kr < 4; kr++) - { - roiHopper += checkRoi+4; - checkRoi = rd[i].rBuf->S[roiHopper]; - if (checkRoi != roi[jr]) - { - factPrintf(kFatal, 1, "Inconsistent Roi accross patches B=%d P=%d, expected %d, got %d", kr, jr, roi[jr], checkRoi); -// goto EndBuf; - continue; - } - } - } - //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; - - const int 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 - //rd[i].bufTyp = 0; //ready to read next header - //rd[i].bufLen = frst_len; - //rd[i].bufPos = 0; - continue; - } - if (evID < 0) { //no space left, retry later -#ifdef EVTDEBUG - if (rd[i].bufLen != 0) { - factPrintf(kFatal, 1, "something screwed up"); - } -#endif - rd[i].bufTyp = -1; - rd[i].bufLen = 0; - rd[i].bufPos = rd[i].fadLen; - - //xwait.tv_sec = 0; - //xwait.tv_nsec = 10000000; // sleep for ~10 msec - //nanosleep (&xwait, NULL); - continue; - //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) { - factPrintf(kFatal, 1, "ERROR OVERWRITE %d %d on port %d", xchk, rd[i].fadLen, i); - - uint iq; - for (iq = 0; iq < rd[i].fadLen; iq++) { - if (rd[i].rBuf->B[iq] != rd[i].xBuf->B[iq]) { - factPrintf(kFatal, 1, "ERROR %4d %4d %x %x", i, iq, rd[i].rBuf->B[iq], rd[i].xBuf->B[iq]); - } - } - } -#endif - const int boardId = b; - const int fadBoard = rd[i].rBuf->S[12]; - const int fadCrate = fadBoard / 256; - - if (boardId != (fadCrate * 10 + fadBoard % 256)) { - factPrintf(kWarn, 301, "Board ID mismatch. Expected %d, got %d (C=%d, B=%d)", - boardId, fadBoard, fadCrate, fadBoard % 256); - } - if (mBuffer[evID].board[boardId] != -1) { - factPrintf(kWarn, 501, "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].fadLen - 2], - rd[i].rBuf->B[rd[i].fadLen - 1]); -// goto EndBuf; //--> skip Board - //rd[i].bufTyp = 0; //ready to read next header - //rd[i].bufLen = frst_len; - //rd[i].bufPos = 0; - continue; - } - - const int iDx = evtIdx[evID]; //index into evtCtrl - - memcpy (&mBuffer[evID].FADhead[boardId].start_package_flag, - &rd[i].rBuf->S[0], sizeof(PEVNT_HEADER)); - - int src = sizeof(PEVNT_HEADER) / 2; - for (int px = 0; px < 9; px++) { //different sort in FAD board..... - for (int drs = 0; drs < 4; drs++) { - // pixH = rd[i].rBuf->S[src++]; // ID - src++; - const int pixC = rd[i].rBuf->S[src++]; // start-cell - const int pixR = rd[i].rBuf->S[src++]; // roi -//here we should check if pixH is correct .... - - const int pixS = boardId * 36 + drs * 9 + px; - src++; - - - mBuffer[evID].fEvent->StartPix[pixS] = pixC; - - int dest = pixS * roi[0]; - memcpy (&mBuffer[evID].fEvent->Adc_Data[dest], - &rd[i].rBuf->S[src], roi[0] * 2); - src += pixR; - - if (px == 8) { - const int 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); - } 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; - - mBuffer[evID].nBoard++; - - // have we already reported first (partial) event of this run ??? - if (mBuffer[evID].nBoard>0 && mBuffer[evID].runNum != actrun) - { - actrun = mBuffer[evID].runNum; - - for (int ir = 0; ir < MAX_RUN; ir++) - { - if (runCtrl[ir].runId == actrun) + if (++runCtrl[ir].lastEvt == 0) { - if (++runCtrl[ir].lastEvt == 0) - { - gotNewRun (actrun, NULL); - factPrintf(kInfo, 1, "gotNewRun called for run %d, event %d", - mBuffer[evID].runNum, mBuffer[evID].evNum); - break; - } + gotNewRun (actrun, NULL); + factPrintf(kInfo, 1, "gotNewRun called for run %d, event %d", + mBuffer[evID].runNum, mBuffer[evID].evNum); + break; } } } - - //complete event read ---> flag for next processing - if (mBuffer[evID].nBoard >= actBoards) - { - evtCtrl.evtStat[iDx] = 99; - gi.evtTot++; - } - - -/* - if (mBuffer[evID].nBoard >= actBoards) - { - // have we already reported first event of this run ??? - if (mBuffer[evID].runNum != actrun) - { - actrun = mBuffer[evID].runNum; - - for (int ir = 0; ir < MAX_RUN; ir++) - { - if (runCtrl[ir].runId == actrun) - { - if (++runCtrl[ir].lastEvt == 0) - { - gotNewRun (actrun, mBuffer[evID].FADhead); - factPrintf(kInfo, 1, "gotNewRun called for run %d, event %d", - mBuffer[evID].runNum, mBuffer[evID].evNum); - break; - } - } - } - } - //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 - { -#ifdef EVTDEBUG - debugRead (i, jrd, 0, 0, 0, -2, tsec, tusec); // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet -#endif - continue; - } - - //check if startflag correct; else shift block .... - int k; - 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 = sizeof(PEVNT_HEADER); - 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); -#ifdef EVTDEBUG - memcpy (&rd[i].xBuf->B[0], &rd[i].xBuf->B[k], - rd[i].bufPos); -#endif - } - - if (rd[i].bufPos < minLen) - { -#ifdef EVTDEBUG - debugRead (i, jrd, 0, 0, 0, -2, tsec, tusec); // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet -#endif - 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]; - 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); - factPrintf(kDebug, 1, "sk %3d head: %5d %5d %5d %10d %4d %6d", - i, rd[i].fadLen, rd[i].evtID, rd[i].ftmID, rd[i].runID, fadboard, jrd); -#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 - } //end interpreting last read -} - } //end of successful read anything - } //finished trying to read all sockets - -#ifdef EVTDEBUG - factDebug(kDebug, -1, "Loop ---- %3d --- %8d", numokx, jrdx); -#endif + } + + //complete event read ---> flag for next processing + if (mBuffer[evID].nBoard >= actBoards) + { + evtCtrl.evtStat[iDx] = 99; + gi.evtTot++; + } + } + else + { + //we are reading event header + + //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++) + { + if (rd[i].rBuf->B[k] == start.B[1] && rd[i].rBuf->B[k+1] == start.B[0]) + break; + } + rd[i].skip += k; + + //no start of header found + if (k >= rd[i].bufPos - 1) + { + rd[i].bufPos = 0; + rd[i].bufLen = sizeof(PEVNT_HEADER); + 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]; + rd[i].bufTyp = 1; //ready to read full record + rd[i].bufLen = rd[i].fadLen - rd[i].bufPos; + + if (rd[i].runID == 0) + rd[i].runID = g_actTime; + + const 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, tv.tv_sec, tv.tv_usec); // i=socket; jrd=#bytes; ievt=eventid;-1=start event + } // end if data or header + } // end for loop over all sockets gi.numRead[numok]++; @@ -1834,9 +1687,9 @@ } - - gj.deltaT = 1000; //temporary, must be improved - - for (int ib = 0; ib < NBOARDS; ib++) - gj.totBytes[ib] += gj.rateBytes[ib]; + gj.deltaT = 1000; //temporary, must be improved + + for (int ib = 0; ib < NBOARDS; ib++) + gj.totBytes[ib] += gj.rateBytes[ib]; + #ifdef THOMAS_MALLOC gj.totMem = tgb_memory; @@ -1844,73 +1697,65 @@ gj.totMem = g_maxMem; #endif - 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; -/* - } - 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 = 1000; // sleep for ~1 usec - } - nanosleep (&xwait, NULL); - } - */ - } //and do next loop over all sockets ... + + 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; - 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 ... - factPrintf(kInfo, -1, "Close all sockets..."); - for (int 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 (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 .... + 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 ... + factPrintf(kInfo, -1, "Close all sockets..."); + + for (int i = 0; i < MAX_SOCK; i++) + { + if (rd[i].sockStat != 0) + continue; + + GenSock(-1, i, 0, NULL, &rd[i]); //close and destroy open socket + if (i%7) + continue; + + 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 + } } @@ -1970,8 +1815,9 @@ } - if (gi_reset > 0) { - if (gi_resetW > 0) { + if (gi_reset > 0) + { + if (gi_resetW > 0) CloseRunFile (0, 0, 0); //ask all Runs to be closed - } + if (gi_resetX > 0) { @@ -1987,6 +1833,4 @@ } - - factPrintf(kInfo, -1, "Exit read Process..."); @@ -1997,6 +1841,8 @@ gi_runStat = -99; gj.readStat = -99; + factStat (gj); factStatNew (gi); + return 0; @@ -2376,7 +2222,4 @@ //close run for whatever reason - if (runCtrl[j].runId == gi_myRun) - gi_myRun = g_actTime; - int ii = 0; if (cond)