Changeset 15494 for trunk/FACT++/src

Timestamp:

05/04/13 11:22:54 (12 years ago)

Author:

tbretz

Message:

Reverting to last revision.

File:

• trunk/FACT++/src/EventBuilder.c (modified) (42 diffs)

trunk/FACT++/src/EventBuilder.c

@@ -73 +73 @@
 
 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 +166 @@
     // 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 +220 @@
 /*-----------------------------------------------------------------*/
 
+
+
+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 +438 @@
     }
 
-    // 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<MAX_RUN; k++)
     {
         // Check if run already registered (old entries should have runId==-1)
-        if (runCtrl[idx].roi0 != nRoi[0] || runCtrl[idx].roi8 != nRoi[8])
+        if (runCtrl[k].runId == runID)
         {
-            factPrintf(kError, 931, "Mismatch of roi within run. Expected roi=%d and roi_tm=%d, got %d and %d (runID=%d, evID=%d)",
-                       runCtrl[idx].roi0, runCtrl[idx].roi8, nRoi[0], nRoi[8], runID, evID);
-            return -9301;
+            // FIXME: Compare to previous event
+            if (runCtrl[k].roi0 != nRoi[0] || runCtrl[k].roi8 != nRoi[8])
+            {
+                factPrintf(kError, 931, "Mismatch of roi within run. Expected roi=%d and roi_tm=%d, got %d and %d (runID=%d, evID=%d)",
+                           runCtrl[k].roi0, runCtrl[k].roi8, nRoi[0], nRoi[8], runID, evID);
+                return -9301;
+            }
+
+            found = 1;
+            break;
+        }
+
+        // This is just for sanity. We use the oldest free entry (until
+        // we have understood the concept and can use "just" a free entry
+        if (runCtrl[k].runId==0 && runCtrl[k].closeTime < oldest)
+        {
+            oldest = runCtrl[k].closeTime;
+            jold = k;
         }
     }
 
-    // If there is none in the queue, we have to search for the correct entry
-    if (evtCtrl_frstPtr==evtCtrl_lastPtr)
+    if (!found) // Run not yet registered, register run
     {
-        idx = -1;
-
-        for (int k=0; k<MAX_RUN; k++)
-        {
-            if (runCtrl[k].runId==runID)
-            {
-                idx = k;
-                break;
-            }
-        }
-    }
-
-    struct timeval tv;
-    gettimeofday(&tv, NULL);
-
-    const uint32_t tsec = tv.tv_sec;
-    const uint32_t tusec = tv.tv_usec;
-
-    // Run not yet registered, register run
-    // If we haven't found a corresponding entry in the queue, or the runId has changed
-    // we find the oldest empty entry in the runCtrl array and create a new runCtrl entry
-    if (idx<0 || runCtrl[idx].runId!=runID)
-    {
-        // If there is none in the queue or
-        idx = -1;
-
-        uint oldest = g_actTime + 1000;
-        for (int k=0; k<MAX_RUN; k++)
-        {
-            // This is just for sanity. We use the oldest free entry (until
-            // we have understood the concept and can use "just" a free entry
-            if (runCtrl[k].runId==0 && runCtrl[k].closeTime < oldest)
-            {
-                oldest = runCtrl[k].closeTime;
-                idx = k;
-            }
-        }
-
-        if (idx<0)
+        if (jold < 0)
         {
             factPrintf(kFatal, 883, "Not able to register the new run %d", runID);
@@ -478 +490 @@
         }
 
-        factPrintf(kInfo, 503, "New run %d (evt=%d, idx=%d) registered with roi=%d and roi_tm=%d",
-                   runID, evID, idx, nRoi[0], nRoi[8]);
-
-        runCtrl[idx].runId      = runID;
-        runCtrl[idx].roi0       = nRoi[0];  // FIXME: Make obsolete!
-        runCtrl[idx].roi8       = nRoi[8];  // FIXME: Make obsolete!
-        runCtrl[idx].fileId     = -2;
-        runCtrl[idx].lastEvt    = 1;         // Number of events partially started to read
-        runCtrl[idx].actEvt     = 0;         // Number of written events (write)
-        runCtrl[idx].procEvt    = 0;         // Number of successfully checked events (checkEvent)
-        runCtrl[idx].maxEvt     = 999999999; // max number events allowed
-        runCtrl[idx].lastTime   = tsec;      // Time when the last event was written
-        runCtrl[idx].closeTime  = tsec + 3600 * 24;     //max time allowed
+        const int evFree = jold;
+
+        factPrintf(kInfo, 503, "New run %d (evID=%d, evFree=%d) registered with roi=%d and roi_tm=%d",
+                   runID, evID, evFree, nRoi[0], nRoi[8]);
+
+        runCtrl[evFree].runId      = runID;
+        runCtrl[evFree].roi0       = nRoi[0];  // FIXME: Make obsolete!
+        runCtrl[evFree].roi8       = nRoi[8];  // FIXME: Make obsolete!
+        runCtrl[evFree].fileId     = -2;
+        runCtrl[evFree].procId     = -2;
+        runCtrl[evFree].lastEvt    = 1;         // Number of events partially started to read
+        runCtrl[evFree].actEvt     = 0;         // Number of written events (write)
+        runCtrl[evFree].procEvt    = 0;         // Number of successfully checked events (checkEvent)
+        runCtrl[evFree].maxEvt     = 999999999; // max number events allowed
+        runCtrl[evFree].lastTime   = tsec;      // Time when the last event was written
+        runCtrl[evFree].closeTime  = tsec + 3600 * 24;     //max time allowed
     }
 
@@ -501 +516 @@
         evtCtrl[k].board[b] = -1;
 
-    evtCtrl[k].runCtrl_idx = idx;
     evtCtrl[k].pcTime[0] = tsec;
     evtCtrl[k].pcTime[1] = tusec;
@@ -568 +582 @@
 mBufFree (int i)
 {
+//delete entry [i] from mBuffer:
+//(and make sure multiple calls do no harm ....)
+
    TGB_free(evtCtrl[i].FADhead);
 
@@ -580 +597 @@
    gj.bufTot--;
 
+   /*if (gi_memStat < 0) {
+      if (gj.usdMem <= 0.75 * gj.maxMem)
+         gi_memStat = +1;
+   }*/
+
    return 0;
 
 } /*-----------------------------------------------------------------*/
+
+/*
+void
+resetEvtStat ()
+{
+   for (int i = 0; i < MAX_SOCK; i++)
+      gi.numRead[i] = 0;
+
+   for (int i = 0; i < NBOARDS; i++) {
+      gi.gotByte[i] = 0;
+      gi.gotErr[i] = 0;
+
+   }
+
+   gi.evtGet = 0;               //#new Start of Events read
+   gi.evtTot = 0;               //#complete Events read
+   gi.evtErr = 0;               //#Events with Errors
+   gi.evtSkp = 0;               //#Events incomplete (timeout)
+
+   gi.procTot = 0;              //#Events processed
+   gi.procErr = 0;              //#Events showed problem in processing
+   gi.procTrg = 0;              //#Events accepted by SW trigger
+   gi.procSkp = 0;              //#Events rejected by SW trigger
+
+   gi.feedTot = 0;              //#Events used for feedBack system
+   gi.feedErr = 0;              //#Events rejected by feedBack
+
+   gi.wrtTot = 0;               //#Events written to disk
+   gi.wrtErr = 0;               //#Events with write-error
+
+   gi.runOpen = 0;              //#Runs opened
+   gi.runClose = 0;             //#Runs closed
+   gi.runErr = 0;               //#Runs with open/close errors
+
+   return;
+}*/ /*-----------------------------------------------------------------*/
 
 uint64_t reportIncomplete(int id, const char *txt)
@@ -704 +762 @@
     READ_STRUCT rd[NBOARDS];       //buffer to read IP and afterwards store in mBuffer
 
-    uint32_t actrun = 0;
-
    const int minLen = sizeof(PEVNT_HEADER);  //min #bytes needed to check header: full header for debug
 
+   //start.S = 0xFB01;
+   //stop.S = 0x04FE;
 
 /* initialize run control logics */
@@ -713 +771 @@
       runCtrl[i].runId  =  0;
       runCtrl[i].fileId = -2;
-   }
-
-   int gi_reset, gi_resetR, gi_resetS, gi_resetW, gi_resetX;
+      runCtrl[i].procId = -2;
+   }
    gi_resetS = gi_resetR = 9;
 
@@ -727 +784 @@
    //time in seconds
    uint gi_SecTime = time(NULL);;
-   g_actTime = gi_SecTime;
 
    const int cntsock = 8 - NUMSOCK ;
@@ -777 +833 @@
    while (g_runStat >= 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 +849 @@
 
           gi_NumConnect[b] = 0;       //must close all connections
+          //gi.numConn[b] = 0;
           gj.numConn[b] = 0;
 
@@ -856 +920 @@
               continue;
 
+          //numok++;
+
           if (rd[i].bufLen>0)
           {
@@ -867 +933 @@
                   // 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 +951 @@
                   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 +1073 @@
               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 +1171 @@
           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 +1200 @@
           // 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<MAX_RUN; ir++)
           {
@@ -1180 +1254 @@
       const int count = (evtCtrl_lastPtr-evtCtrl_frstPtr+MAX_EVT)%MAX_EVT;
 
-      // This could be improved having the pointer which separates the queue with
-      // the incomplete events from the queue with the complete events
       for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT)
       {
@@ -1190 +1262 @@
 
               // Event has not yet timed out or was reported already
-              if (evtCtrl[k0].evtStat==90 || evtCtrl[k0].pcTime[0]>=g_actTime - 30)
+              if (evtCtrl[k0].evtStat==90 || evtCtrl[k0].pcTime[0]/*evtCtrl[k0].lastRecv*/>=g_actTime - 30)
                   continue;
 
@@ -1224 +1296 @@
               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 +1352 @@
 
        //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 +1393 @@
                evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % MAX_EVT;
                evtCtrl[k0].evtStat = -1;
-           }
+          }
 
            usleep(1);
@@ -1341 +1421 @@
    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 +1444 @@
         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 +1504 @@
    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 +1540 @@
    {
        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 +1565 @@
            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; j++)
+                   if (runCtrl[j].runId == irun)
+                       break;
+
+               if (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<MAX_RUN; j++)
+               {
+                   if (runCtrl[j].fileId == 0)
+                   {
+                       runCtrl[j].procId = 2; //--> 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 +1616 @@
                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 +1674 @@
            }
 
-           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 +1693 @@
        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 +1716 @@
    }
 
+   /*
+   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 +1786 @@
         ii |= 8; // = 4;
 
+    if (runCtrl[j].procId == 0)
+    {
+        runFinish1(runCtrl[j].runId);
+        runCtrl[j].procId = 92;
+    }
+
     runCtrl[j].closeTime = g_actTime - 1;
 
@@ -1642 +1816 @@
 /* *** 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 +1841 @@
        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; j++)
+                   if (runCtrl[j].runId == irun)
+                       break;
+
+               if (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<MAX_RUN;j++)
+           {
+               if (runCtrl[j].runId == actrun)
+               {
+                   lastStartedTime = runCtrl[j].lastTime;
+                   runIdFound = 1;
+               }
+           }
+       }
+
+      if (runIdFound == 0)
+      {
+          factPrintf(kInfo, 0, "An Active run (number %u) has been registered, but it could not be found in the runs list", actrun);
+      }
+
+       //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 (int j=0;j<MAX_RUN;j++)
+       {
+           if ((runCtrl[j].fileId < 0) &&
+               (runCtrl[j].lastTime < lastStartedTime) &&
+               (runCtrl[j].runId != 0))
+           {
+               factPrintf(kInfo, 0, "writeEvt: No file will be opened for run %u. Last run: %u (started)", runCtrl[j].runId, actrun);
+               ;//TODO notify that this run will never be opened
+           }
+       }
+       */
 
        // Although the are no pending events, we have to check if a run should be closed (timeout)
@@ -1714 +1994 @@
        }
 
-       usleep(1);
+      //seems we have nothing to do, so sleep a little
+      if (numWrite == 0)
+          usleep(1);
 
        //nothing left to do
@@ -1720 +2002 @@
        {
            factPrintf(kInfo, -1, "Finish Write Process ...");
+           gw_runStat = -22;      //==> we should exit
            gj.writStat = -22;     //==> we should exit
            break;
        }
 
+       gw_runStat = gi_runStat;
        gj.writStat = gj.readStat;
    }
@@ -1734 +2018 @@
    }
 
+   gw_runStat = -99;
    gj.writStat = -99;
 
@@ -1752 +2037 @@
    struct timespec xwait;
 
+   gi_runStat = gp_runStat = gw_runStat = 0;
    gj.readStat = gj.procStat = gj.writStat = 0;
 
@@ -1761 +2047 @@
       runCtrl[i].fileId = -2;
    }
+
+//prepare for subProcesses
+   gi_maxSize = g_maxSize;
+   if (gi_maxSize <= 0)
+      gi_maxSize = 1;
 
    gi_maxProc = g_maxProc;
@@ -1969 +2260 @@
 
    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;