Merge pull request #195 from sy-c/master

v2.4.2

Author: sy-c

doc/releaseNotes.md

@@ -359,3 +359,9 @@ This file describes the main feature changes for each readout.exe released versi
 - Updated configuration parameters:
   - added readout.disableTimefarmes:  when set, all timeframe-related features are disabled (STF slicing, TF rate limits, etc). All data are tagged with TF id = 0. To be used for some calibration runs not using a central trigger clock.
   - added consumer-FMQchannel.checkResources: controls which resources are checked for fitting unmanaged region. This is a comma-separated list of items to be checked. By default, no checks are done. Recommended value: /dev/shm, MemAvailable.
+
+## v2.4.2 - 28/06/2021
+- verbose logs auto-mute (aggregator, consumer-FMQchannel).
+- consumer-FMQchannel: drop TF on error (to avoid unhappy STFB when sending incomplete data, eg on "data page too small" or "no page left" conditions).
+- added memory pool usage statistics (to help tuning buffer pages count and size).
+- added some ZeroMQ options for consumerZMQ and equipmentZMQ.

src/ConsumerFMQchannel.cxx

@@ -14,8 +14,10 @@
 #include "MemoryPagesPool.h"
 #include "ReadoutStats.h"
 #include "ReadoutUtils.h"
+#include "CounterStats.h"
 #include <atomic>
 #include <chrono>
+#include <inttypes.h>
 
 #ifdef WITH_FAIRMQ
 
@@ -105,6 +107,8 @@ class ConsumerFMQchannel : public Consumer
   int memoryPoolPageSize;
   int memoryPoolNumberOfPages;
 
+  CounterStats repackSizeStats; // keep track of page size used when repacking
+
  public:
   std::vector<FairMQMessagePtr> messagesToSend; // collect HBF messages of each update
   uint64_t messagesToSendSize;                  // size (bytes) of messagesToSend payload
@@ -277,6 +281,12 @@ class ConsumerFMQchannel : public Consumer
 
   ~ConsumerFMQchannel()
   {
+    // log memory pool statistics
+    if (mp!=nullptr) {
+      theLog.log(LogInfoDevel_(3003), "Consumer %s - memory pool statistics ... %s", name.c_str(), mp->getStats().c_str());
+      theLog.log(LogInfoDevel_(3003), "Consumer %s - STFB repacking statistics ... number: %" PRIu64 " average page size: %" PRIu64 " max page size: %" PRIu64, name.c_str(), repackSizeStats.getCount(), (uint64_t)repackSizeStats.getAverage(), repackSizeStats.getMaximum());
+    }
+    
     // release in reverse order
     mp = nullptr;
     memoryBuffer = nullptr; // warning: data range may still be referenced in memory bank manager
@@ -596,11 +606,16 @@ class ConsumerFMQchannel : public Consumer
         for (auto& f : pendingFrames) {
           totalSize += f.HBlength;
         }
+	
+	// keep stats on repack page size
+	repackSizeStats.set(totalSize);
+	
         // allocate
         // todo: same code as for header -> create func/lambda
         // todo: send empty message if no page left in buffer
         if (memoryPoolPageSize < totalSize) {
-          theLog.log(LogWarningSupport_(3230), "page size too small %d < %d", memoryPoolPageSize, totalSize);
+	  static InfoLogger::AutoMuteToken token(LogWarningSupport_(3230));
+          theLog.log(token, "page size too small %d < %d", memoryPoolPageSize, totalSize);
           throw __LINE__;
         }
         DataBlockContainerReference copyBlock = nullptr;
@@ -609,7 +624,8 @@ class ConsumerFMQchannel : public Consumer
         } catch (...) {
         }
         if (copyBlock == nullptr) {
-          theLog.log(LogWarningSupport_(3230), "no page left");
+	  static InfoLogger::AutoMuteToken token(LogWarningSupport_(3230));
+          theLog.log(token, "no page left");
           throw __LINE__;
         }
         auto blockRef = new DataBlockContainerReference(copyBlock);
@@ -705,8 +721,10 @@ class ConsumerFMQchannel : public Consumer
         }
       }
       pendingFrames.clear();
-
-      theLog.log(LogErrorSupport_(3233), "ConsumerFMQ : error %d", err);
+      static InfoLogger::AutoMuteToken token(LogErrorSupport_(3233));
+      theLog.log(token, "ConsumerFMQ : error %d", err);
+      // cleanup buffer, and start fresh (in particular: avoid sending empty message parts)
+      messagesToSend.clear();
       totalPushError++;
       return -1;
     }

src/ConsumerZmq.cxx

@@ -26,10 +26,16 @@ class ConsumerZMQ : public Consumer
   uint64_t nBytesSent = 0;
   uint64_t nBlocksSent = 0;
   std::string cfgAddress = "tcp://127.0.0.1:50001";
-  int cfgZmqMaxQueue = 10;
-  int cfgZmqConflate = 1;
-  int cfgZmqSendTimeout = 1000;
-  int cfgZmqLinger = 1000;
+
+  // default ZMQ settings for data monitoring
+  // settings for CTP readout: ZMQ_CONFLATE=0,ZMQ_IO_THREADS=4,ZMQ_SNDHWM=1000
+  int cfg_ZMQ_CONFLATE = 1; // buffer last message only
+  int cfg_ZMQ_IO_THREADS = 1; // number of IO threads
+  int cfg_ZMQ_LINGER = 1000; // close timeout
+  int cfg_ZMQ_SNDBUF = 16*1024*1024; // kernel transmit buffer size                
+  int cfg_ZMQ_SNDHWM = 10; // max send queue size
+  int cfg_ZMQ_SNDTIMEO = 1000; // send timeout
+
   double cfgMaxRate = 0; // max number of pages per second (average)
   int cfgPagesPerBurst = 1; // number of pages per burst (peak successive pages accepted without avg rate check)
   int pagesInBurst = 0; // current number of pages in burst
@@ -50,8 +56,37 @@ class ConsumerZMQ : public Consumer
     if (cfgPagesPerBurst < 1) {
       cfgPagesPerBurst = 1;
     }
-    theLog.log(LogInfoDevel_(3002), "ZeroMQ server @ %s, rate limit = %.4f pages/s, in burst of %d pages", cfgAddress.c_str(), cfgMaxRate, cfgPagesPerBurst);
 
+    // configuration parameter: | consumer-zmq-* | zmqOptions | string |  | Additional ZMQ options, as a comma-separated list of key=value pairs. Possible keys: ZMQ_CONFLATE, ZMQ_IO_THREADS, ZMQ_LINGER, ZMQ_SNDBUF, ZMQ_SNDHWM, ZMQ_SNDTIMEO. |
+    std::string cfg_ZMQOptions = "";
+    cfg.getOptionalValue<std::string>(cfgEntryPoint + ".zmqOptions", cfg_ZMQOptions);
+    std::map<std::string, std::string> mapOptions;
+    if (getKeyValuePairsFromString(cfg_ZMQOptions, mapOptions)) {
+      throw("Can not parse configuration item zmqOptions");
+    }
+    bool isOk=1;
+    for (auto& it : mapOptions) {
+      int value = atoi(it.second.c_str());
+      if (it.first=="ZMQ_CONFLATE") { cfg_ZMQ_CONFLATE = value; }
+      else if (it.first=="ZMQ_IO_THREADS") { cfg_ZMQ_IO_THREADS = value; }
+      else if (it.first=="ZMQ_LINGER") { cfg_ZMQ_LINGER = value; }
+      else if (it.first=="ZMQ_SNDBUF") { cfg_ZMQ_SNDBUF = value; }
+      else if (it.first=="ZMQ_SNDHWM") { cfg_ZMQ_SNDHWM = value; }
+      else if (it.first=="ZMQ_SNDTIMEO") { cfg_ZMQ_SNDTIMEO = value; }
+      else {
+        theLog.log(LogErrorSupport_(3102), "Wrong ZMQ option %s", it.first.c_str());
+	isOk = 0;
+        continue;
+      }
+    }
+    if (!isOk) {
+      throw __LINE__;
+    }	
+  
+    // log config summary
+    theLog.log(LogInfoDevel_(3002), "ZeroMQ PUB server @ %s, rate limit = %.4f pages/s, in burst of %d pages", cfgAddress.c_str(), cfgMaxRate, cfgPagesPerBurst);
+    theLog.log(LogInfoDevel_(3002), "ZMQ options: ZMQ_SNDHWM=%d ZMQ_CONFLATE=%d ZMQ_SNDTIMEO=%d ZMQ_LINGER=%d ZMQ_SNDBUF=%d ZMQ_IO_THREADS=%d", cfg_ZMQ_SNDHWM, cfg_ZMQ_CONFLATE, cfg_ZMQ_SNDTIMEO, cfg_ZMQ_LINGER, cfg_ZMQ_SNDBUF, cfg_ZMQ_IO_THREADS);
+  
     int linerr = 0;
     int zmqerr = 0;
     for (;;) {
@@ -61,25 +96,26 @@ class ConsumerZMQ : public Consumer
         zmqerr = zmq_errno();
         break;
       }
+      
+      zmq_ctx_set(context, ZMQ_IO_THREADS, cfg_ZMQ_IO_THREADS);
+      if (zmq_ctx_get(context, ZMQ_IO_THREADS) != cfg_ZMQ_IO_THREADS) {
+        linerr = __LINE__;
+        break;
+      }
       zh = zmq_socket(context, ZMQ_PUB);
-      /*
       if (zh==nullptr) { linerr=__LINE__; zmqerr=zmq_errno(); break; }
-      int timeout = 1000;
-      zmqerr=zmq_setsockopt(zh, ZMQ_RCVTIMEO, (void*) &timeout, sizeof(int));
+      zmqerr = zmq_bind(zh, cfgAddress.c_str());
       if (zmqerr) { linerr=__LINE__; break; }
-      int linger = 0;
-      zmqerr=zmq_setsockopt(zh, ZMQ_LINGER, (void*) &linger, sizeof(int));
+      zmqerr = zmq_setsockopt(zh, ZMQ_CONFLATE, &cfg_ZMQ_CONFLATE, sizeof(cfg_ZMQ_CONFLATE));
+      if (zmqerr) { linerr=__LINE__; break; }
+      zmqerr = zmq_setsockopt(zh, ZMQ_LINGER, (void*)&cfg_ZMQ_LINGER, sizeof(cfg_ZMQ_LINGER));
+      if (zmqerr) { linerr=__LINE__; break; }
+      zmqerr = zmq_setsockopt(zh, ZMQ_SNDBUF, (void*)&cfg_ZMQ_SNDBUF, sizeof(cfg_ZMQ_SNDBUF));
+      if (zmqerr) { linerr=__LINE__; break; }
+      zmqerr=zmq_setsockopt(zh, ZMQ_SNDHWM, &cfg_ZMQ_SNDHWM, sizeof(cfg_ZMQ_SNDHWM));
+      if (zmqerr) { linerr=__LINE__; break; }
+      zmqerr = zmq_setsockopt(zh, ZMQ_SNDTIMEO, (void*)&cfg_ZMQ_SNDTIMEO, sizeof(cfg_ZMQ_SNDTIMEO));
       if (zmqerr) { linerr=__LINE__; break; }
-      */
-      zmqerr = zmq_bind(zh, cfgAddress.c_str());
-      if (zmqerr) {
-        linerr = __LINE__;
-        break;
-      }
-      zmq_setsockopt(zh, ZMQ_SNDHWM, &cfgZmqMaxQueue, sizeof(cfgZmqMaxQueue)); // max queue size
-      zmq_setsockopt(zh, ZMQ_CONFLATE, &cfgZmqConflate, sizeof(cfgZmqConflate)); // buffer last message only
-      zmq_setsockopt(zh, ZMQ_SNDTIMEO, (void*)&cfgZmqSendTimeout, sizeof(cfgZmqSendTimeout)); // send timeout
-      zmq_setsockopt(zh, ZMQ_LINGER, (void*)&cfgZmqLinger, sizeof(cfgZmqLinger)); // close timeout
       break;
     }
 
@@ -103,6 +139,7 @@ class ConsumerZMQ : public Consumer
     }
     // the stats are not meaningfull for a ZMQ PUB: send always works...
     // theLog.log(LogInfoDevel_(3003), "ZeroMQ stats: %" PRIu64 "/%" PRIu64 " blocks sent/discarded", nBlocksSent, nBlocksDropped);
+    theLog.log(LogInfoDevel_(3003), "ZeroMQ publish stats: %" PRIu64 " blocks %" PRIu64 " bytes", nBlocksSent, nBytesSent);
   }
 
   int pushData(DataBlockContainerReference& b)

src/DataBlockAggregator.cxx

@@ -170,7 +170,8 @@ Thread::CallbackResult DataBlockAggregator::executeCallback()
         uint64_t tfId = db->header.timeframeId;
         uint64_t sourceId = (((uint64_t)db->header.equipmentId) << 32) | ((uint64_t)db->header.linkId);
         if (tfId <= lastTimeframeId) {
-          theLog.log(LogWarningSupport_(3235), "Discarding late data for TF %" PRIu64 " (source = 0x%" PRIx64 ")", tfId, sourceId);
+	  static InfoLogger::AutoMuteToken token(LogWarningSupport_(3004));
+          theLog.log(token, "Discarding late data for TF %" PRIu64 " (source = 0x%" PRIx64 ")", tfId, sourceId);
         } else {
           tStf& stf = stfBuffer[tfId];
           stf.tfId = tfId;
@@ -221,7 +222,12 @@ Thread::CallbackResult DataBlockAggregator::executeCallback()
           nSources = it->second.sstf.size(); // keep track of number of sources in first TF
         }
         nStfPushed++;
-        lastTimeframeId = it->second.tfId;
+	uint64_t newTimeframeId = it->second.tfId;
+	if (newTimeframeId > lastTimeframeId + 1) {
+	  static InfoLogger::AutoMuteToken token(LogWarningSupport_(3004));
+          theLog.log(token, "Gap in timeframe ids detected: previous = %" PRIu64 " new = %" PRIu64, lastTimeframeId, newTimeframeId);
+	}
+        lastTimeframeId = newTimeframeId;
         /*
         if (lastTimeframeId % 10 == 1) {
           theLog.log(LogDebugTrace, "LastTimeframeId=%lu deltaT=%f",lastTimeframeId,tmax-tmin);
@@ -259,7 +265,8 @@ int DataBlockSlicer::appendBlock(DataBlockContainerReference const& block, doubl
 
   if (sourceId.linkId != undefinedLinkId) {
     if (sourceId.linkId >= maxLinks) {
-      theLog.log(LogWarningSupport_(3004), "wrong link id %d > %d", sourceId.linkId, maxLinks - 1);
+      static InfoLogger::AutoMuteToken token(LogWarningSupport_(3004));
+      theLog.log(token, "wrong link id %d > %d", sourceId.linkId, maxLinks - 1);
       return -1;
     }
   }

src/MemoryPagesPool.cxx

@@ -140,6 +140,9 @@ MemoryPagesPool::~MemoryPagesPool()
 
 void* MemoryPagesPool::getPage()
 {
+  // update statistics
+  poolStats.set((CounterValue)getNumberOfPagesAvailable());
+
   // get a page from fifo, if available
   void* ptr = nullptr;
   pagesAvailable->pop(ptr);
@@ -260,3 +263,7 @@ bool MemoryPagesPool::isPageValid(void* pagePtr)
 }
 
 size_t MemoryPagesPool::getDataBlockMaxSize() { return pageSize - headerReservedSpace; }
+
+std::string MemoryPagesPool::getStats() {
+  return "number of pages used: " + std::to_string(poolStats.getTotal()) + " average free pages: " + std::to_string((uint64_t)poolStats.getAverage()) + " minimum free pages: " + std::to_string(poolStats.getMinimum());
+}

src/MemoryPagesPool.h

@@ -16,6 +16,7 @@
 #include <functional>
 #include <map>
 #include <memory>
+#include <string>
 
 #include "CounterStats.h"
 #include "DataBlockContainer.h"
@@ -69,6 +70,8 @@ class MemoryPagesPool
 
   bool isPageValid(void* page); // check to see if a page address is valid
 
+  std::string getStats(); // return a string summarizing memory pool usage statistics
+
  private:
   std::unique_ptr<AliceO2::Common::Fifo<void*>> pagesAvailable; // a buffer to keep track of individual pages
 
@@ -104,6 +107,8 @@ class MemoryPagesPool
   // t3: releasepage->getpage
   // t4: getpage->releasepage
   CounterStats t1, t2, t3, t4;
+  
+  CounterStats poolStats; // keep track of number of free pages in the pool 
 };
 
 #endif // #ifndef _MEMORYPAGESPOOL_H

src/ReadoutEquipment.cxx

@@ -253,6 +253,10 @@ void ReadoutEquipment::stop()
 
 ReadoutEquipment::~ReadoutEquipment()
 {
+  if (mp != nullptr) {
+    theLog.log(LogInfoDevel_(3003), "Equipment %s - memory pool statistics ... %s", name.c_str(), mp->getStats().c_str());
+  }
+	
   // check if mempool still referenced
   if (!mp.unique()) {
     theLog.log(LogInfoDevel_(3008), "Equipment %s :  mempool still has %d references\n", name.c_str(), (int)mp.use_count());
@@ -339,7 +343,7 @@ Thread::CallbackResult ReadoutEquipment::threadCallback(void* arg)
 	}
 
 	// handle RDH-formatted data
-	if (ptr->isRdhEquipment) {
+	if (ptr->cfgRdhUseFirstInPageEnabled) {
           ptr->processRdh(nextBlock);
 	}