Changeset 16618

Timestamp:

06/03/13 11:10:01 (12 years ago)

Author:

tbretz

Message:

Added a different way to count the byted for the statistics; added the possibility to choose 'read always the board which has transmitted the least number of bytes first'

File:

• trunk/FACT++/src/EventBuilder.cc (modified) (14 diffs)

trunk/FACT++/src/EventBuilder.cc

@@ -7 +7 @@
 #include <cstdarg>
 #include <list>
+#include <queue>
 
 #include <boost/algorithm/string/join.hpp>
@@ -25 +26 @@
 //#define USE_SELECT
 //#define COMPLETE_EPOLL
+//#define PRIORITY_QUEUE
 
 // Reading only 1024: 13:  77Hz, 87%
@@ -207 +209 @@
 
     timeval  time;
-    uint64_t rateBytes;
-    uint32_t skip;     // number of bytes skipped before start of event
+    uint64_t totBytes;  // total received bytes
+    uint64_t relBytes;  // total released bytes
+    uint32_t skip;      // number of bytes skipped before start of event
 
     uint32_t len() const { return uint32_t(H.package_length)*2; }
@@ -217 +220 @@
     // --------------------------------
 
-    READ_STRUCT() : socket(-1), connected(false), rateBytes(0)
+    READ_STRUCT() : socket(-1), connected(false), totBytes(0), relBytes(0)
     {
         if (fd_epoll<0)
@@ -231 +234 @@
     bool check(int, sockaddr_in addr);
     bool read();
+
 };
+
+#ifdef PRIORITY_QUEUE
+struct READ_STRUCTcomp
+{
+    bool operator()(const READ_STRUCT *r1, const READ_STRUCT *r2)
+    {
+        const int64_t rel1 = r1->totBytes - r1->relBytes;
+        const int64_t rel2 = r2->totBytes - r2->relBytes;
+        return rel1 > rel2;
+    }
+};
+#endif
 
 int READ_STRUCT::wait()
@@ -382 +398 @@
     }
 
-    bufPos = B;  // no byte read so far
-    skip   = 0;       // start empty
+    bufPos   = B;  // no byte read so far
+    skip     = 0;  // start empty
+    totBytes = 0;
+    relBytes = 0;
 
     factPrintf(MessageImp::kInfo, "Connected socket %d (%d)", sockId, socket);
@@ -427 +445 @@
     }
 
-    rateBytes += jrd;
+    totBytes += jrd;
 
     // are we skipping this board ...
@@ -1076 +1094 @@
 #endif
 
-#if defined(USE_POLL)
+#ifdef PRIORITY_QUEUE
+        priority_queue<READ_STRUCT*, vector<READ_STRUCT*>, READ_STRUCTcomp> prio;
+
+        for (int i=0; i<NBOARDS; i++)
+            if (rd[i].connected)
+                prio.push(rd+i);
+
+        if (!prio.empty()) do
+#endif
+
+
+#ifdef USE_POLL
         for (int jj=0; jj<nn; jj++)
 #endif
-#if defined(USE_EPOLL)
+#ifdef USE_EPOLL
         for (int jj=0; jj<rc_epoll; jj++)
 #endif
-#if !defined(USE_EPOLL) && !defined(USE_POLL)
+#if !defined(USE_EPOLL) && !defined(USE_POLL) && !defined(PRIORITY_QUEUE)
         for (int jj=0; jj<NBOARDS; jj++)
 #endif
         {
+#ifdef PRIORITY_QUEUE
+            READ_STRUCT *rs = prio.top();
+#endif
 #ifdef USE_SELECT
             if (!FD_ISSET(rs->socket, &readfs))
@@ -1106 +1138 @@
 #endif
 
-#if !defined(USE_POLL) && !defined(USE_EPOLL)
+#if !defined(USE_POLL) && !defined(USE_EPOLL) && !defined(PRIORITY_QUEUE)
             const int i = (jj%4)*10 + (jj/4);
             READ_STRUCT *rs = &rd[i];
@@ -1265 +1297 @@
                     primaryQueue.post(evt);
 
+                // package_len is 0 if nothing was received.
+                for (int ib=0; ib<40; ib++)
+                    rd[ib].relBytes += (*it)->FADhead[ib].package_length;
+
                 it++;
                 evtCtrl.pop_front();
@@ -1299 +1335 @@
 #endif
         } // end for loop over all sockets
+#ifdef PRIORITY_QUEUE
+        while (0); // convert continue into break ;)
+#endif
 
         // ==================================================================
@@ -1334 +1373 @@
             const uint64_t rep = reportIncomplete(evt, "timeout");
             factReportIncomplete(rep);
+
+            // package_len is 0 when nothing was received from this board
+            for (int ib=0; ib<40; ib++)
+                rd[ib].relBytes += (*it)->FADhead[ib].package_length;
 
             it++;
@@ -1354 +1397 @@
         bool changed = false;
 
+        static vector<uint64_t> store(NBOARDS);
+
         for (int ib=0; ib<NBOARDS; ib++)
         {
-            gj.rateBytes[ib]  = rd[ib].rateBytes;
-            gj.totBytes[ib]  += rd[ib].rateBytes;
+            gj.rateBytes[ib] = store[ib]>rd[ib].totBytes ? rd[ib].totBytes : rd[ib].totBytes-store[ib];
+            gj.relBytes[ib]  = rd[ib].totBytes-rd[ib].relBytes;
+
+            store[ib] = rd[ib].totBytes;
 
             if (rd[ib].check(g_port[ib].sockDef, g_port[ib].sockAddr))
@@ -1369 +1416 @@
 
         Memory::max_inuse = 0;
-
-        for (int ib=0; ib<NBOARDS; ib++)
-            rd[ib].rateBytes = 0;
 
         // =================================================================