Changeset 15483

Timestamp:

05/03/13 22:30:18 (12 years ago)

Author:

tbretz

Message:

Replaced MAX_EVT*MAX_RUN by just MAX_EVT; simplified evtStat to have only the minimum amount of states (easier to read); remove incomplete events which are superseeded by complete events (this might happen if a new run was started)

Location:

trunk/FACT++/src

Files:

• EventBuilder.c (modified) (28 diffs)
• FAD.h (modified) (1 diff)

trunk/FACT++/src/EventBuilder.c

@@ -69 +69 @@
 int CloseRunFile (uint32_t runId, uint32_t closeTime, uint32_t maxEvt);
 
-int evtCtrl_frstPtr;
-int evtCtrl_lastPtr;
+int evtCtrl_frstPtr;   // First event in queue
+int evtCtrl_lastPtr;   // pointer to next free slot
 
 int g_maxProc;
@@ -105 +105 @@
 GUI_STAT gj;
 
-EVT_CTRL evtCtrl[MAX_EVT * MAX_RUN]; //control of events during processing
+#define MAX_EVT   65536  // ( 300s @ 220Hz; 16GB = 5000 evt @ roi=1024 (27s) ; 18000 evt @ roi = 300 )
+#define MAX_RUN       8  // Number of concurrent runs
+
+EVT_CTRL evtCtrl[MAX_EVT]; //control of events during processing
 
 void factPrintf(int severity, int id, const char *fmt, ...)
@@ -377 +380 @@
 int mBufEvt(const READ_STRUCT *rs)
 {
-// generate a new Event into mBuffer:   
-// make sure only complete Event are possible, so 'free' will always work
-// returns index into mBuffer[], or negative value in case of error
-// error: <-9000 if roi screwed up (not consistent with run)
-//        <-8000                   (not consistent with event)
-//        <-7000                   (not consistent with board)
-//        < 0    if no space left
-
     int nRoi[9];
     if (!checkRoiConsistency(rs->rBuf, nRoi))
@@ -395 +390 @@
     const int  fadNum = rs->evtID;
 
-    const int beg = (evtCtrl_lastPtr + MAX_EVT*MAX_RUN - 1) % (MAX_EVT*MAX_RUN);
-    const int end = (evtCtrl_frstPtr + MAX_EVT*MAX_RUN - 1) % (MAX_EVT*MAX_RUN);
-
-    for (int k=beg; k!=end; k=(k+MAX_EVT*MAX_RUN-1)%(MAX_EVT*MAX_RUN))
+    const int beg = (evtCtrl_lastPtr + MAX_EVT - 1) % MAX_EVT;
+    const int end = (evtCtrl_frstPtr + MAX_EVT - 1) % MAX_EVT;
+
+    for (int k=beg; k!=end; k=(k+MAX_EVT-1)%MAX_EVT)
     {
         // If the run is different, go on searching.
@@ -437 +432 @@
 
     // Check if the control structure still has space left
-    if (end-beg==1 || (end==0 && beg==MAX_EVT*MAX_RUN-1))
+    if (end-beg==1 || (end==0 && beg==MAX_EVT-1))
     {
         factPrintf(kError, 881, "No control slot to keep event %d (run %d) %d %d", evID, runID, beg, end);
@@ -537 +532 @@
     evtCtrl[k].FADhead   = NULL;
 
+    //    -1:   kInValid
+    //     0:   kValid
+    //  1-40:   kIncomplete
+    //    90:   kIncompleteReported
+    //   100:   kCompleteEventInBuffer
+    //  1000+x: kToBeProcessedByThreadX
+    //  5000:   kToBeWritten
+    // 10000:   kToBeDeleted
+
     evtCtrl[k].evtStat = 0;
 
@@ -543 +547 @@
     // in two instructions. Must be done only _after_ the contents
     // have been initialized
-    evtCtrl_lastPtr = (evtCtrl_lastPtr+1) % (MAX_EVT*MAX_RUN);
+    evtCtrl_lastPtr = (evtCtrl_lastPtr+1) % MAX_EVT;
 
     return k;
@@ -638 +642 @@
 }*/ /*-----------------------------------------------------------------*/
 
-void reportIncomplete(int id)
-{
-    factPrintf(kWarn, 601, "%5d skip incomplete evt %8d",
-               evtCtrl[id].evNum, id);
+uint64_t reportIncomplete(int id, const char *txt)
+{
+    factPrintf(kWarn, 601, "skip incomplete evt (run=%d, evt=%d, %s)",
+               evtCtrl[id].runNum, evtCtrl[id].evNum, txt);
 
     uint64_t report = 0;
@@ -660 +664 @@
         }
 
-        // FIXME: Is that really 'b' or should that be 'ib' ?
+        // FIXME: This is not synchronous... it reports
+        // accoridng to the current connection status, not w.r.t. to the
+        // one when the event was taken.
         if (gi_NumConnect[ib]<=0) // board not connected
         {
@@ -677 +683 @@
     factOut(kWarn, 601, str);
 
-    factReportIncomplete(report);
+    return report;
 }
 
@@ -808 +814 @@
 
       // initialize mBuffer (mark all entries as unused\empty)
-      for (int i = 0; i < MAX_EVT * MAX_RUN; i++)
+      for (int i = 0; i < MAX_EVT; i++)
       {
           evtCtrl[i].evNum = evtCtrl[i].nRoi = -1;
@@ -1201 +1207 @@
           }
 
+          const int beg = (idx             + MAX_EVT - 1) % MAX_EVT;
+          const int end = (evtCtrl_frstPtr + MAX_EVT - 1) % MAX_EVT;
+
+          // we have just completed an event... so all previous events
+          // must have been completed already. If they are not, there
+          // is no need to wait for the timeout, because they will never
+          // get completed. We can just ensure that if we check for the previous
+          // event to be complete every time we receive a new complete event.
+          // If we find an incomplete one, we remove all consecutive
+          // incomplete ones.
+          for (int k=beg; k!=end; k=(k+MAX_EVT-1)%MAX_EVT)
+          {
+              // We are done if we find a complete or fully processed event
+              if (evtCtrl[k].evtStat>=100)
+                  break;
+
+              // we do not call factReportIncomplete here, because by starting a new
+              // run and having received the first complete event from that run
+              // the user has expressed that the old events are obsolste now
+              // and the run will be closed anyway
+              if (evtCtrl[k].evtStat>=0 && evtCtrl[k].evtStat<90)
+              {
+                  reportIncomplete(k, "expired");
+                  evtCtrl[k].evtStat = 90;
+              }
+          }
+
           // Flag that the event is ready for processing
-          evtCtrl[idx].evtStat = 99;
+          evtCtrl[idx].evtStat = 100;
 
       } // end for loop over all sockets
@@ -1219 +1252 @@
       //delete those that are written to disk ....
 
-      const int count = (evtCtrl_lastPtr-evtCtrl_frstPtr+MAX_EVT*MAX_RUN)%(MAX_EVT*MAX_RUN);
-
-      for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
+      const int count = (evtCtrl_lastPtr-evtCtrl_frstPtr+MAX_EVT)%MAX_EVT;
+
+      for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
       {
           // Check the more likely case first: incomplete events
-          if (evtCtrl[k0].evtStat>=0 && evtCtrl[k0].evtStat<92)
+          if (evtCtrl[k0].evtStat>=0 && evtCtrl[k0].evtStat<100)
           {
               gj.bufNew++;     //incomplete event in Buffer
 
               // Event has not yet timed out or was reported already
-              if (evtCtrl[k0].evtStat>=90 || evtCtrl[k0].pcTime[0]/*evtCtrl[k0].lastRecv*/>=g_actTime - 30)
+              if (evtCtrl[k0].evtStat==90 || evtCtrl[k0].pcTime[0]/*evtCtrl[k0].lastRecv*/>=g_actTime - 30)
                   continue;
 
@@ -1235 +1268 @@
               // It doesn't matter if that takes comparably long,
               // because we have to stop the run anyway.
-              reportIncomplete(k0);
+              const uint64_t rep = reportIncomplete(k0, "timeout");
+              factReportIncomplete(rep);
 
               //timeout for incomplete events
-              evtCtrl[k0].evtStat = 91;
+              evtCtrl[k0].evtStat = 90;
               gj.evtSkip++;
 
               continue;
           }
-
-          // complete event in Buffer
-          //if (evtCtrl[k0].evtStat >= 95)
-          //    gj.bufEvt++;
 
           // Check the less likely case: 'useless' or 'delete'
           // evtState==0 can happen if the event was initialized (some data received)
           // but the data did not make sense (e.g. inconsistent rois)
-          if (evtCtrl[k0].evtStat==0 || evtCtrl[k0].evtStat >= 9000)
+          if (evtCtrl[k0].evtStat==0 || evtCtrl[k0].evtStat == 10000)
           {
               mBufFree(k0);   //event written--> free memory
               evtCtrl[k0].evtStat = -1;
 
-              if (k0==evtCtrl_frstPtr)
-                  evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % (MAX_EVT*MAX_RUN);
-              else
-                  factPrintf(kError, -1, "Freed a non-first slot");
-
               gj.evtWrite++;
               gj.rateWrite++;
-
+          }
+
+          // Remove leading invalidated slots from queue
+          if (evtCtrl[evtCtrl_frstPtr].evtStat==-1 && k0==evtCtrl_frstPtr)
+          {
+              evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % MAX_EVT;
               continue;
           }
@@ -1271 +1301 @@
           // first complete events which processing is pending, we can stop.
           // All other events (if everything works well) must have the same state.
-          break;
+          // FIXME: This only works if we start from lastPtr and go down to frstPtr
+          //break;
       }
 
@@ -1353 +1384 @@
 
            // flag incomplete events as 'read finished'
-           if (evtCtrl[k0].evtStat>0 && evtCtrl[k0].evtStat < 90)
-               evtCtrl[k0].evtStat = 91;
-
-           if (evtCtrl[k0].evtStat==0 || evtCtrl[k0].evtStat>900)
+           // We cannot just detele all events, because some might currently being processed,
+           // so we have to wait until the processing thread currently processing the event
+           // signals that the event can be deleted. (Note, that there are currently never
+           // two threads processing the same event at the same time)
+           if ((evtCtrl[k0].evtStat>0 && evtCtrl[k0].evtStat<90) || evtCtrl[k0].evtStat==10000)
            {
                mBufFree(k0);   //event written--> free memory
+               evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % MAX_EVT;
                evtCtrl[k0].evtStat = -1;
-
-               evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % (MAX_EVT*MAX_RUN);
-           }
+          }
 
            usleep(1);
@@ -1410 +1441 @@
     {
         int numWait = 0;
-        int numProc = 0;
-
-        for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
+
+        for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
         {
             if (evtCtrl[k0].evtStat != 1000 + threadID)
@@ -1422 +1452 @@
             }
 
-            int jret = 9100; // flag to be deleted (gi_resetR>1 : flush buffers asap)
-
-            if (gi_resetR<=1)
+            /*
+            // gi_resetR>1 : flush buffers asap
+            if (gi_resetR>1)
             {
-                jret = subProcEvt(threadID, evtCtrl[k0].FADhead,
-                                  evtCtrl[k0].fEvent, NULL/*mBuffer[id].buffer*/);
-
-
-                if (jret <= threadID) {
-                    factPrintf(kError, -1, "Process %ld wants to send event to process %d... not allowed.", threadID, jret);
-                    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[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)
+                factPrintf(kError, -1, "Process %ld wants to send event to process %d... not allowed.", threadID, jret);
+
+            if (jret<=threadID)
+            {
+                evtCtrl[k0].evtStat = 10000;            // flag as 'to be deleted'
+                continue;
             }
 
-            evtCtrl[k0].evtStat = jret;
-            numProc++;
+            if (jret>=gi_maxProc)
+            {
+                evtCtrl[k0].evtStat = 5000;            // flag as 'to be written'
+                continue;
+            }
+
+            evtCtrl[k0].evtStat = 1000 + jret;       // flag for next proces
         }
 
@@ -1450 +1487 @@
         }
 
-        //seems we have nothing to do, so sleep a little
-        if (numProc == 0)
-            usleep(1);
+        usleep(1);
     }
 
@@ -1507 +1542 @@
        int numProc = 0;
 
-       for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
+       for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
        {
-           if (evtCtrl[k0].evtStat <= 90 || evtCtrl[k0].evtStat >= 1000)
+           if (evtCtrl[k0].evtStat<90 || evtCtrl[k0].evtStat>=1000)
            {
-               if (evtCtrl[k0].evtStat >= 0 && evtCtrl[k0].evtStat< 90)
+               if (/*evtCtrl[k0].evtStat>=0 &&*/ evtCtrl[k0].evtStat<90)
                    numWait++;
 
@@ -1517 +1552 @@
            }
 
+           /*
            //we are asked to flush buffers asap
            if (gi_resetR > 1)
            {
-               evtCtrl[k0].evtStat = 9991;
+               evtCtrl[k0].evtStat = 9000;
                continue;
-           }
+           }*/
 
            //-------- it is better to open the run already here, so call can be used to initialize
@@ -1611 +1647 @@
            if (runCtrl[lastRun].procId != 0)
            {
-               evtCtrl[k0].evtStat = 9091;
+               evtCtrl[k0].evtStat = 10000; // flag 'to be deleted'
                continue;
            }
@@ -1645 +1681 @@
            if (rc < 0)
            {
-               evtCtrl[k0].evtStat = 9999;        //flag event to be skipped
+               evtCtrl[k0].evtStat = 10000;        // flag event to be deleted
                //gi.procErr++;
            }
            else
            {
-               evtCtrl[k0].evtStat = 1000;
+               evtCtrl[k0].evtStat = 1000;       // flag 'start processing'
                runCtrl[lastRun].procEvt++;
            }
@@ -1680 +1716 @@
    }
 
-   for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
+   /*
+   for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
    {
        if (evtCtrl[k0].evtStat >= 0 && evtCtrl[k0].evtStat < 1000)
-           evtCtrl[k0].evtStat = 9800;    //flag event as 'processed'
-   }
+           evtCtrl[k0].evtStat = 9000;    //flag event as 'processed'
+   }*/
 
    gp_runStat = -99;
@@ -1800 +1837 @@
        int numWait  = 0;
 
-       for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
+       // Note that the current loop does not at all gurantee that
+       // the events are written in the correct order.
+       for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
        {
-           if (evtCtrl[k0].evtStat <= 5000 || evtCtrl[k0].evtStat >= 9000)
+           if (evtCtrl[k0].evtStat<5000 || evtCtrl[k0].evtStat >= 10000)
            {
-               if (evtCtrl[k0].evtStat > 0 && evtCtrl[k0].evtStat < 9000)
+               if (/*evtCtrl[k0].evtStat > 0 &&*/ evtCtrl[k0].evtStat < 5000)
                    numWait++;
 
@@ -1811 +1850 @@
 
            //we must drain the buffer asap
+           /*
            if (gi_resetR > 1)
            {
-               evtCtrl[k0].evtStat = 9904;
+               evtCtrl[k0].evtStat = 9000;
                continue;
-           }
+           }*/
 
            const uint32_t irun = evtCtrl[k0].runNum;
@@ -1868 +1908 @@
            }
 
+           /*
            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 = 9903;
-           }
+               //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
@@ -1891 +1931 @@
                    runCtrl[lastRun].lastTime = g_actTime;
                    runCtrl[lastRun].actEvt++;
-
-                   evtCtrl[k0].evtStat = 9901;
                }
                else
-               {
                    factPrintf(kError, 503, "writeEvt: Writing event for run %d failed (runWrite)", irun);
-                   evtCtrl[k0].evtStat = 9902;
-               }
 
                checkAndCloseRun(lastRun, irun, rc<0, 1);
            }
+
+           evtCtrl[k0].evtStat = 10000;  // event written (or has to be discarded) -> delete
        }
 /*

trunk/FACT++/src/FAD.h

@@ -177 +177 @@
 
 //---------------------------------------------------------------
-
-#define MAX_RUN     8
-#define MAX_EVT   32768  //don't worry, for events is MAX_RUN*MAX_EVT
 
 typedef void* FileHandle_t ;