Index: trunk/FACT++/src/EventBuilder.c =================================================================== --- trunk/FACT++/src/EventBuilder.c (revision 15493) +++ trunk/FACT++/src/EventBuilder.c (revision 15494) @@ -73,17 +73,34 @@ 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; + +uint32_t actrun = 0; + uint gi_NumConnect[NBOARDS]; //4 crates * 10 boards -//EVT_STAT gi; +EVT_STAT gi; GUI_STAT gj; @@ -149,6 +166,4 @@ // Add the last free slot to the stack TGB_entry *entry = (TGB_entry*)malloc(sizeof(TGB_entry)); - - // FIXME: Really free memory if memory usuage exceeds g_maxMem entry->prev = tgb_last; @@ -205,4 +220,18 @@ /*-----------------------------------------------------------------*/ + + +int +runFinish1 (uint32_t runnr) +{ + factPrintf(kInfo, 173, "Should finish(1) run %d (but not yet possible)", runnr); + return 0; +} +int +runFinish (uint32_t runnr) +{ + factPrintf(kInfo, 173, "Should finish run %d (but not yet possible)", runnr); + return 0; +} int @@ -409,68 +438,51 @@ } - // If we have already queued at least one event, - // check the roi if the previous event - // already belongs to the same run. - - // Get the runCtrl entry of the previous event. - // If none is in the queue (we are anyhow super fast) - // just get the correct entry from the runCtrl array. - int idx = evtCtrl[beg].runCtrl_idx; - - // If there is an event in the queue and it has the same runID, we can use - // that event to check for the roi consistency throughout the run - if (evtCtrl_frstPtr!=evtCtrl_lastPtr && runCtrl[idx].runId==runID) + // FIXME: This should be the time of the first receiped board + struct timeval tv; + gettimeofday (&tv, NULL); + + const uint32_t tsec = tv.tv_sec; + const uint32_t tusec = tv.tv_usec; + + //check if runId already registered in runCtrl + + uint oldest = g_actTime + 1000; + int jold = -1; + + int found = 0; + + // fileId==-2: not yet used or run assigned but not open + // fileId== 0: file open + // fileId>0: run closed + + for (int k=0; k= 0 && g_reset == 0) { + gi_runStat = g_runStat; gj.readStat = g_runStat; + + struct timeval tv; + gettimeofday (&tv, NULL); + g_actTime = tv.tv_sec; + g_actUsec = tv.tv_usec; + for (int b = 0; b < NBOARDS; b++) @@ -786,4 +849,5 @@ gi_NumConnect[b] = 0; //must close all connections + //gi.numConn[b] = 0; gj.numConn[b] = 0; @@ -856,4 +920,6 @@ continue; + //numok++; + if (rd[i].bufLen>0) { @@ -867,7 +933,11 @@ // 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; } @@ -881,9 +951,13 @@ GenSock(s0, i, 0, NULL, &rd[i]); + //gi.gotErr[b]++; + gi_NumConnect[i]-= cntsock ; + //gi.numConn[b]--; gj.numConn[i]--; continue; } + // Success (jrd > 0) gj.rateBytes[i] += jrd; @@ -999,4 +1073,5 @@ rd[i].rBuf->B[rd[i].fadLen - 2] != 0x04) { + //gi.evtErr++; factPrintf(kError, 301, "End-of-event flag wrong on socket %3d for event %4d (len=%5d), got %3d %3d", i, rd[i].evtID, rd[i].fadLen, @@ -1096,5 +1171,5 @@ if (evtCtrl[idx].nBoard==1 && evtCtrl[idx].runNum != actrun) { - // Signal the fadctrl that a new run has been started + // Signal the fadctrl that a new run has been started gotNewRun(evtCtrl[idx].runNum, NULL); @@ -1125,6 +1200,5 @@ // This is a non-ideal hack to lower the probability that // in mBufEvt the search for correct entry in runCtrl - // will not return a super-old entry. I don't want - // to manipulate that in another thread. + // will not return a super-old entry for (int ir=0; ir=g_actTime - 30) + if (evtCtrl[k0].evtStat==90 || evtCtrl[k0].pcTime[0]/*evtCtrl[k0].lastRecv*/>=g_actTime - 30) continue; @@ -1224,4 +1296,11 @@ continue; } + + // The first events in the queue are either incomplete or + // can be deleted (processing finished). As soon as we reach the + // first complete events which processing is pending, we can stop. + // All other events (if everything works well) must have the same state. + // FIXME: This only works if we start from lastPtr and go down to frstPtr + //break; } @@ -1273,4 +1352,5 @@ //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 } @@ -1313,5 +1393,5 @@ evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % MAX_EVT; evtCtrl[k0].evtStat = -1; - } + } usleep(1); @@ -1341,8 +1421,9 @@ factPrintf(kInfo, -1, "%ld Bytes flagged as in-use.", tgb_inuse); + gi_runStat = -99; gj.readStat = -99; factStat (gj); - //factStatNew (gi); + factStatNew (gi); return 0; @@ -1363,24 +1444,23 @@ for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) { - // This is a threading issue... the evtStat might have been invalid - // but the frstPtr is not yet updated - if (evtCtrl[k0].evtStat==-1) - continue; - - // If we find the first event still waiting for processing - // there will be only unprocessed events after this one in the queue - if (evtCtrl[k0].evtStat<1000+threadID) - { - numWait = 1; - break; - } - - // If the event was processed already, skip it - // We could replace that to a moving pointer pointing to the first - // non-processed event - if (evtCtrl[k0].evtStat!=1000+threadID) - continue; - - const int jret = subProcEvt(threadID, evtCtrl[k0].FADhead, evtCtrl[k0].fEvent, 0); + if (evtCtrl[k0].evtStat != 1000 + threadID) + { + if (evtCtrl[k0].evtStat < 1000 + threadID) + numWait++; + + continue; + } + + /* + // gi_resetR>1 : flush buffers asap + if (gi_resetR>1) + { + evtCtrl[k].evtStat = 9000; // flag as 'to be deleted' + continue; + + }*/ + + const int jret = subProcEvt(threadID, evtCtrl[k0].FADhead, + evtCtrl[k0].fEvent, NULL/*mBuffer[id].buffer*/); if (jret>0 && jret<=threadID) @@ -1424,5 +1504,28 @@ int status; + 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) + factPrintf(kInfo, -1, "Starting process-thread with %d subprocesses", gi_maxProc); + +/* 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]; @@ -1437,25 +1540,23 @@ { int numWait = 0; + int numProc = 0; for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) { - // This is a threading issue... the evtStat might have been invalid - // but the frstPtr is not yet updated - if (evtCtrl[k0].evtStat==-1) + if (evtCtrl[k0].evtStat<90 || evtCtrl[k0].evtStat>=1000) + { + if (/*evtCtrl[k0].evtStat>=0 &&*/ evtCtrl[k0].evtStat<90) + numWait++; + continue; - - // If we find the first incomplete event which is not supposed to - // be processed, there are only more incomplete events in the queue - if (evtCtrl[k0].evtStat<90) + } + + /* + //we are asked to flush buffers asap + if (gi_resetR > 1) { - numWait = 1; - break; - } - - // If the event was processed already, skip it. - // We could replace that to a moving pointer pointing to the first - // non-processed event - if (evtCtrl[k0].evtStat>=1000) + evtCtrl[k0].evtStat = 9000; continue; + }*/ //-------- it is better to open the run already here, so call can be used to initialize @@ -1464,15 +1565,42 @@ const int32_t ievt = evtCtrl[k0].evNum; - const int idx = evtCtrl[k0].runCtrl_idx; - if (runCtrl[idx].runId!=irun) + // Find entry in runCtrl which belongs to the event mBuffer[id] + // (only check if there is a need to check) + if (runCtrl[lastRun].runId != irun) { - //factPrintf(kFatal, 901, "procEvt: runCtrl entry for run %d vanished (evt=%d)", irun, ievt); - // FIXME: What is the right action? (Flag event for deletion?) - continue; + //check which fileID to use (or open if needed) + int j; + for (j=0;j=MAX_RUN) + { + factPrintf(kFatal, 901, "writeEvt: Can not find run %d for event %d in %d", irun, ievt, k0); + // FIXME: What is the right action? (Flag event for deletion?) + continue; + } + + lastRun = j; } // File not yet open - if (runCtrl[idx].fileId < 0) + if (runCtrl[lastRun].fileId < 0) { + //---- we need to open a new run ==> make sure all older runs are + //---- finished and marked to be closed .... + // This loop is unique to procEvt + for (int j=0; j do no longer accept events for processing + + //---- problem: processing still going on ==> must wait for closing .... + factPrintf(kInfo, -1, "procEvt: Finished run since new one opened %d", runCtrl[j].runId); + runFinish1(runCtrl[j].runId); + } + } + RUN_HEAD actRun; actRun.Version = 1; @@ -1488,15 +1616,37 @@ memcpy(actRun.FADhead, evtCtrl[k0].FADhead, NBOARDS*sizeof(PEVNT_HEADER)); - runCtrl[idx].fileHd = runOpen(irun, &actRun, sizeof (actRun)); - if (runCtrl[idx].fileHd == NULL) + runCtrl[lastRun].fileHd = runOpen (irun, &actRun, sizeof (actRun)); + if (runCtrl[lastRun].fileHd == NULL) { - factPrintf(kError, 502, "procEvt: Could not open new file for run %d (idx=%d, evt=%d, runOpen failed)", irun, idx, ievt); - runCtrl[idx].fileId = 91; + factPrintf(kError, 502, "procEvt: Could not open a file for run %d (runOpen failed)", irun); + runCtrl[lastRun].fileId = 91; + runCtrl[lastRun].procId = 91; // Is not set in writeEvt continue; } - runCtrl[idx].fileId = 0; - - factPrintf(kInfo, -1, "procEvt: Opened new file for run %d (idx=%d, evt=%d)", irun, idx, ievt); + runCtrl[lastRun].fileId = 0; + runCtrl[lastRun].procId = 0; // Is not set in writeEvt + + factPrintf(kInfo, -1, "procEvt: Opened new file for run %d (evt=%d)", irun, ievt); + } + + //-------- also check if run shall be closed (==> skip event, but do not close the file !!! ) + if (runCtrl[lastRun].procId == 0) + { + if (runCtrl[lastRun].closeTime < g_actTime || + runCtrl[lastRun].lastTime < g_actTime - 300 || + runCtrl[lastRun].maxEvt <= runCtrl[lastRun].procEvt) + { + factPrintf(kInfo, 502, "procEvt: Reached end of run condition for run %d", irun); + runFinish1 (runCtrl[lastRun].runId); + runCtrl[lastRun].procId = 1; + } + } + + // Skip event because of no active run + if (runCtrl[lastRun].procId != 0) + { + evtCtrl[k0].evtStat = 10000; // flag 'to be deleted' + continue; } @@ -1524,13 +1674,18 @@ } - const int rc = eventCheck(evtCtrl[k0].runNum, evtCtrl[k0].FADhead, evtCtrl[k0].fEvent); + const int rc = eventCheck(evtCtrl[k0].runNum, evtCtrl[k0].FADhead, + evtCtrl[k0].fEvent); + //gi.procTot++; + numProc++; + if (rc < 0) { evtCtrl[k0].evtStat = 10000; // flag event to be deleted + //gi.procErr++; } else { evtCtrl[k0].evtStat = 1000; // flag 'start processing' - runCtrl[idx].procEvt++; + runCtrl[lastRun].procEvt++; } } @@ -1538,11 +1693,16 @@ if (gj.readStat < -10 && numWait == 0) { //nothing left to do factPrintf(kInfo, -1, "Exit Processing Process ..."); + gp_runStat = -22; //==> we should exit gj.procStat = -22; //==> we should exit return 0; } - usleep(1); - + //seems we have nothing to do, so sleep a little + if (numProc == 0) + usleep(1); + + gp_runStat = gi_runStat; gj.procStat = gj.readStat; + } @@ -1556,4 +1716,12 @@ } + /* + for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) + { + if (evtCtrl[k0].evtStat >= 0 && evtCtrl[k0].evtStat < 1000) + evtCtrl[k0].evtStat = 9000; //flag event as 'processed' + }*/ + + gp_runStat = -99; gj.procStat = -99; @@ -1618,4 +1786,10 @@ ii |= 8; // = 4; + if (runCtrl[j].procId == 0) + { + runFinish1(runCtrl[j].runId); + runCtrl[j].procId = 92; + } + runCtrl[j].closeTime = g_actTime - 1; @@ -1642,9 +1816,23 @@ /* *** main loop writing event (including opening and closing run-files */ +// cpu_set_t mask; +// int cpu = 1; //write thread + factPrintf(kInfo, -1, "Starting write-thread"); + +/* 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) { - //int numWrite = 0; + int numWrite = 0; int numWait = 0; @@ -1653,53 +1841,145 @@ for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) { - // This is a threading issue... the evtStat might have been invalid - // but the frstPtr is not yet updated - if (evtCtrl[k0].evtStat==-1) + if (evtCtrl[k0].evtStat<5000 || evtCtrl[k0].evtStat >= 10000) + { + if (/*evtCtrl[k0].evtStat > 0 &&*/ evtCtrl[k0].evtStat < 5000) + numWait++; + continue; - - // If we find the first non-written event which is not supposed to - // be written, there are only more incomplete events in the queue - if (evtCtrl[k0].evtStat<5000) + } + + //we must drain the buffer asap + /* + if (gi_resetR > 1) { - numWait = 1; - break; - } - - // If the event was written already already, skip it - // We could replace that to a moving pointer pointing to the first - // non-processed event - if (evtCtrl[k0].evtStat!=5000) + evtCtrl[k0].evtStat = 9000; continue; + }*/ const uint32_t irun = evtCtrl[k0].runNum; const int32_t ievt = evtCtrl[k0].evNum; - const int idx = evtCtrl[k0].runCtrl_idx; - - if (runCtrl[idx].runId!=irun) + // Find entry in runCtrl which belongs to the event mBuffer[id] + // (only check if there is a need to check) + if (runCtrl[lastRun].runId != irun) { - //factPrintf(kFatal, 901, "writeEvt: runCtrl entry for run %d vanished (evt=%d)", irun, ievt); - // FIXME: What is the right action? (Flag event for deletion?) - continue; + //check which fileID to use (or open if needed) + int j; + for (j=0;j=MAX_RUN) + { + factPrintf(kFatal, 901, "writeEvt: Can not find run %d for event %d in %d", irun, ievt, k0); + // FIXME: What is the right action? + continue; + } + + lastRun = j; } + // File not yet open + if (runCtrl[lastRun].fileId < 0) + { + RUN_HEAD actRun; + actRun.Version = 1; + actRun.RunType = -1; //to be adapted + actRun.Nroi = evtCtrl[k0].nRoi; //runCtrl[lastRun].roi0; + actRun.NroiTM = evtCtrl[k0].nRoiTM; //runCtrl[lastRun].roi8; + actRun.RunTime = evtCtrl[k0].pcTime[0];//runCtrl[lastRun].firstTime; + actRun.RunUsec = evtCtrl[k0].pcTime[1];//runCtrl[lastRun].firstUsec; + actRun.NBoard = NBOARDS; + actRun.NPix = NPIX; + actRun.NTm = NTMARK; + + memcpy(actRun.FADhead, evtCtrl[k0].FADhead, NBOARDS * sizeof (PEVNT_HEADER)); + + runCtrl[lastRun].fileHd = runOpen (irun, &actRun, sizeof (actRun)); + if (runCtrl[lastRun].fileHd == NULL) + { + factPrintf(kError, 502, "writeEvt: Could not open a file for run %d (runOpen failed)", irun); + runCtrl[lastRun].fileId = 91; + continue; + } + + runCtrl[lastRun].fileId = 0; + factPrintf(kInfo, -1, "writeEvt: Opened new file for run %d (evt %d)", irun, ievt); + } + + /* + if (runCtrl[lastRun].fileId > 0) + { + // There is an event but file is already closed + //if (runCtrl[j].fileId < 100) + //{ + // factPrintf(kWarn, 123, "writeEvt: File for run %d is closed", irun); + // runCtrl[j].fileId += 100; + //} + + evtCtrl[k0].evtStat = 9000; + }*/ + // File is open - if (runCtrl[idx].fileId==0) + if (runCtrl[lastRun].fileId==0) { - const int rc = runWrite(runCtrl[idx].fileHd, evtCtrl[k0].fEvent, 0); + const int rc = runWrite(runCtrl[lastRun].fileHd, evtCtrl[k0].fEvent, + 0/*sizeof (evtCtrl[k0])*/); if (rc >= 0) { // Sucessfully wrote event - runCtrl[idx].lastTime = g_actTime; - runCtrl[idx].actEvt++; + runCtrl[lastRun].lastTime = g_actTime; + runCtrl[lastRun].actEvt++; } else factPrintf(kError, 503, "writeEvt: Writing event for run %d failed (runWrite)", irun); - checkAndCloseRun(idx, irun, rc<0, 1); + checkAndCloseRun(lastRun, irun, rc<0, 1); } evtCtrl[k0].evtStat = 10000; // event written (or has to be discarded) -> delete } +/* + //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 + //actrun + uint32_t lastStartedTime = 0; + uint32_t runIdFound = 1; + + //If we have an active run, look for its start time + if (actrun != 0) + { + runIdfound = 0; + for (int j=0;j we should exit gj.writStat = -22; //==> we should exit break; } + gw_runStat = gi_runStat; gj.writStat = gj.readStat; } @@ -1734,4 +2018,5 @@ } + gw_runStat = -99; gj.writStat = -99; @@ -1752,4 +2037,5 @@ struct timespec xwait; + gi_runStat = gp_runStat = gw_runStat = 0; gj.readStat = gj.procStat = gj.writStat = 0; @@ -1761,4 +2047,9 @@ runCtrl[i].fileId = -2; } + +//prepare for subProcesses + gi_maxSize = g_maxSize; + if (gi_maxSize <= 0) + gi_maxSize = 1; gi_maxProc = g_maxProc; @@ -1969,7 +2260,9 @@ g_maxMem = 1024 * 1024; //MBytes +//g_maxMem = g_maxMem * 1024 *10 ; //10GBytes g_maxMem = g_maxMem * 200; //100MBytes g_maxProc = 20; + g_maxSize = 30000; g_runStat = 40;