Merge pull request #190 from sy-c/master

v2.4.0

Author: sy-c

CMakeLists.txt

@@ -67,8 +67,7 @@ find_package(FairMQ)
 find_package(ZLIB)
 find_package(RDMA)
 find_package(Occ)
-#find_package(BookkeepingApiCpp)
-#find_package(cpprestsdk)
+find_package(BookkeepingApiCpp)
 find_package(ZMQ)
 
 # extract include directories from targets
@@ -339,6 +338,12 @@ add_executable(
 	$<TARGET_OBJECTS:objReadoutUtils>
 )
 
+# a test for FMQ memory
+add_executable(
+	o2-readout-test-fmq-memory
+        ${SOURCE_DIR}/testFmqMemory.cxx
+)
+
 
 # a test to check memory banks
 add_executable(
@@ -401,7 +406,7 @@ add_executable(
 
 # disable some executables when corresponding dependencies not found
 if (NOT FairMQ_FOUND)
-	set_target_properties(o2-readout-test-fmq-tx o2-readout-test-fmq-rx o2-readout-test-fmq-perf-tx o2-readout-test-fmq-perf-rx PROPERTIES EXCLUDE_FROM_ALL 1)
+	set_target_properties(o2-readout-test-fmq-tx o2-readout-test-fmq-rx o2-readout-test-fmq-perf-tx o2-readout-test-fmq-perf-rx  o2-readout-test-fmq-memory PROPERTIES EXCLUDE_FROM_ALL 1)
 endif ()
 
 # set include and libraries for all
@@ -420,6 +425,12 @@ if (${ZMQ_FOUND})
   list (APPEND executables o2-readout-monitor)
 endif()
 
+# special minimal build for test-fmq-memory (to avoid grpc garbage from occ when using valgrind)
+foreach (exe o2-readout-test-fmq-memory)
+  list (APPEND executables ${exe})
+  target_include_directories(${exe} PRIVATE ${READOUT_INCLUDE_DIRS})
+  target_link_libraries(${exe} PRIVATE $<$<BOOL:${FairMQ_FOUND}>:FairMQ::FairMQ>)
+endforeach()
 
 ###################################################
 # files installation

doc/releaseNotes.md

@@ -345,3 +345,12 @@ This file describes the main feature changes for each readout.exe released versi
 - consumer-stats: publish/print the current timeframe Id sent to STFB.
 - consumer-stats: ZMQ stats client cleanup timeout, to avoid blocking on exit.
 - auto-mute RDH warnings: verbosity reduced if many successive logs done in a short time.
+
+## v2.4.0 - 03/06/2021
+- equipment-player: fixed bug in "autochunk" replay mode. There was a "last packet invalid" error wrongly reported (and aborting replay) in the rare case when a link change would occur exactly at the beginning of a page.
+- Default value of equipment-rorc.rdhUseFirstInPageEnabled is now 1 for all RDH equipments (RORC, emu, player).
+- FMQ stats not printed when consoleUpdate=1 unless there is a running consumerFMQchannel with disableSending=0.
+- tfRateLimit is handled in the equipment directly and avoid potential issues with timeframe slicing at very slow rates.
+- equipment-cruemulator: TF id extracted from trigger counters (single timer source for improved coherency).
+- Memory allocation policy updated: all readout memory is locked (RAM only, can not be swapped). A warning is reported if not.
+- consumer-FMQchannel: checks are done before FMQ shared memory region is created, to avoid going in a state with over-committed memory (no checks done in FMQ library about the validity of the region created, which can cause severe crash when trying to access it). Both /proc/meminfo (MemFree) and /dev/shm (if using shmem transport type) should report enough available memory before proceeding. Memory is also immediately locked and zeroed to avoid later crashes.

src/ConsumerFMQchannel.cxx

@@ -118,6 +118,8 @@ class ConsumerFMQchannel : public Consumer
     if (cfgDisableSending) {
       theLog.log(LogInfoDevel_(3002), "FMQ message sending disabled");
       disableSending = true;
+    } else {
+      gReadoutStats.isFairMQ = 1; // enable FMQ stats
     }
 
     // configuration parameter: | consumer-FairMQChannel-* | enableRawFormat | int | 0 | If 0, data is pushed 1 STF header + 1 part per HBF. If 1, data is pushed in raw format without STF headers, 1 FMQ message per data page. If 2, format is 1 STF header + 1 part per data page.|
@@ -183,14 +185,47 @@ class ConsumerFMQchannel : public Consumer
     cfg.getOptionalValue<std::string>(cfgEntryPoint + ".unmanagedMemorySize", cfgUnmanagedMemorySize);
     long long mMemorySize = ReadoutUtils::getNumberOfBytesFromString(cfgUnmanagedMemorySize.c_str());
     if (mMemorySize > 0) {
+    
+      // check system resources first, as region creation does not check available memory, so bad crash could occur later      
+      theLog.log(LogInfoDevel_(3002), "Configuring memory buffer %lld MB", (long long)(mMemorySize/1048576LL));
+	    
+      // free system memory
+      unsigned long long freeBytes;
+      if (getStatsFreeMemory(freeBytes)) {
+        theLog.log(LogWarningSupport_(3230), "Can not get stats about system free memory available");
+      } else {
+          theLog.log(LogInfoSupport_(3230), "Stats free memory available: %lld MB", (long long)(freeBytes/1048576LL));
+	  if ((long long)freeBytes < mMemorySize) {
+          theLog.log(LogErrorSupport_(3230), "Not enough system memory available - check /proc/meminfo");
+          throw "ConsumerFMQ: can not allocate shared memory region";
+	} else {
+
+	}
+      }
+            
+      // free SHM memory
+      // check only if transport is of type shmem
+      if (cfgTransportType == "shmem") {
+	if (getStatsFreeSHM(freeBytes)) {
+          theLog.log(LogWarningSupport_(3230), "Can not get stats about shared memory available");
+	} else {
+          theLog.log(LogInfoSupport_(3230), "Stats shared memory available: %lld MB", (long long)(freeBytes/1048576LL));
+          if ((long long)freeBytes < mMemorySize) {
+            theLog.log(LogErrorSupport_(3230), "Not enough shared memory available - check /dev/shm");
+            throw "ConsumerFMQ: can not allocate shared memory region";
+	  }
+	}
+      }
+            
+      theLog.log(LogInfoDevel_(3008), "Creating FMQ unmanaged memory region");
       memoryBuffer = sendingChannel->Transport()->CreateUnmanagedRegion(mMemorySize, [](void* /*data*/, size_t /*size*/, void* hint) { // cleanup callback
         if (hint != nullptr) {
           DataBlockContainerReference* blockRef = (DataBlockContainerReference*)hint;
           //printf("ack hint=%p page %p\n",hint,(*blockRef)->getData());
           decDataBlockStats((*blockRef)->getData());
           delete blockRef;
         }
-      });
+      },"",0,fair::mq::RegionConfig{true, true});  // lock / zero
 
       theLog.log(LogInfoDevel_(3008), "Got FMQ unmanaged memory buffer size %lu @ %p", memoryBuffer->GetSize(), memoryBuffer->GetData());
     }

src/ConsumerStats.cxx

@@ -171,7 +171,9 @@ class ConsumerStats : public Consumer
     if (consoleUpdate) {
       if (deltaT > 0) {
         theLog.log(LogInfoOps_(3003), "Last interval (%.2fs): blocksRx=%llu, block rate=%.2lf, bytesRx=%llu, rate=%s", deltaT, (unsigned long long)counterBlocksDiff, counterBlocksDiff / deltaT, (unsigned long long)counterBytesDiff, NumberOfBytesToString(counterBytesDiff * 8 / deltaT, "b/s", 1000).c_str());
-        theLog.log(LogInfoOps_(3003), "STFB locked pages: current=%llu, release rate=%.2lf Hz, latency=%.3lf s, current TF = %d", nRfmq, rRfmq, avgTfmq, tfidfmq );
+	if (gReadoutStats.isFairMQ) {
+          theLog.log(LogInfoOps_(3003), "STFB locked pages: current=%llu, release rate=%.2lf Hz, latency=%.3lf s, current TF = %d", nRfmq, rRfmq, avgTfmq, tfidfmq );
+	}
       }
     }
 

src/ReadoutEquipment.cxx

@@ -15,7 +15,7 @@
 
 extern tRunNumber occRunNumber;
 
-ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint)
+ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, bool setRdhEquipment)
 {
 
   // example: browse config keys
@@ -28,6 +28,13 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint)
   // configuration parameter: | equipment-* | name | string| | Name used to identify this equipment (in logs). By default, it takes the name of the configuration section, equipment-xxx |
   cfg.getOptionalValue<std::string>(cfgEntryPoint + ".name", name, cfgEntryPoint);
 
+  // change defaults for equipments generating data with RDH
+  if (setRdhEquipment) {
+    theLog.log(LogInfoDevel_(3002), "Equipment %s: generates data with RDH, using specific defaults", name.c_str());
+    isRdhEquipment = true;
+    cfgRdhUseFirstInPageEnabled = 1; // by default, use first RDH in page
+  }
+
   // configuration parameter: | equipment-* | id | int| | Optional. Number used to identify equipment (used e.g. in file recording). Range 1-65535.|
   int cfgEquipmentId = undefinedEquipmentId;
   cfg.getOptionalValue<int>(cfgEntryPoint + ".id", cfgEquipmentId);
@@ -89,6 +96,10 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint)
     this->debugFirstPages = cfgDebugFirstPages;
   }
 
+  // get TF rate from toplevel config
+  cfgTfRateLimit = 0;
+  cfg.getOptionalValue<double>("readout.tfRateLimit", cfgTfRateLimit);
+
   // log config summary
   theLog.log(LogInfoDevel_(3002), "Equipment %s: from config [%s], max rate=%lf Hz, idleSleepTime=%d us, outputFifoSize=%d", name.c_str(), cfgEntryPoint.c_str(), readoutRate, cfgIdleSleepTime, cfgOutputFifoSize);
   theLog.log(LogInfoDevel_(3008), "Equipment %s: requesting memory pool %d pages x %d bytes from bank '%s', block aligned @ 0x%X, 1st page offset @ 0x%X", name.c_str(), (int)memoryPoolNumberOfPages, (int)memoryPoolPageSize, memoryBankName.c_str(), (int)cfgBlockAlign, (int)cfgFirstPageOffset);
@@ -105,7 +116,7 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint)
   cfg.getOptionalValue<int>(cfgEntryPoint + ".rdhDumpErrorEnabled", cfgRdhDumpErrorEnabled);
   // configuration parameter: | equipment-* | rdhDumpWarningEnabled | int | 0 | If set, a log message is printed for each RDH header warning found.|
   cfg.getOptionalValue<int>(cfgEntryPoint + ".rdhDumpWarningEnabled", cfgRdhDumpWarningEnabled);
-  // configuration parameter: | equipment-* | rdhUseFirstInPageEnabled | int | 0 | If set, the first RDH in each data page is used to populate readout headers (e.g. linkId).|
+  // configuration parameter: | equipment-* | rdhUseFirstInPageEnabled | int | 0 or 1 | If set, the first RDH in each data page is used to populate readout headers (e.g. linkId). Default is 1 for  equipments generating data with RDH, 0 otherwsise. |
   cfg.getOptionalValue<int>(cfgEntryPoint + ".rdhUseFirstInPageEnabled", cfgRdhUseFirstInPageEnabled);
   theLog.log(LogInfoDevel_(3002), "RDH settings: rdhCheckEnabled=%d rdhDumpEnabled=%d rdhDumpErrorEnabled=%d rdhDumpWarningEnabled=%d rdhUseFirstInPageEnabled=%d", cfgRdhCheckEnabled, cfgRdhDumpEnabled, cfgRdhDumpErrorEnabled, cfgRdhDumpWarningEnabled, cfgRdhUseFirstInPageEnabled);
 
@@ -194,6 +205,10 @@ void ReadoutEquipment::start()
   }
   clk0.reset();
 
+  // reset TF rate clock
+  TFregulator.init(cfgTfRateLimit);
+  throttlePendingBlock = nullptr;
+  
   // reset stats timer
   consoleStatsTimer.reset(cfgConsoleStatsUpdateTime * 1000000);
 
@@ -214,6 +229,9 @@ void ReadoutEquipment::stop()
   this->finalCounters();
   ReadoutEquipment::finalCounters();
 
+  // cleanup
+  throttlePendingBlock = nullptr;
+
   for (int i = 0; i < (int)EquipmentStatsIndexes::maxIndex; i++) {
     if (equipmentStats[i].getCount()) {
       theLog.log(LogInfoDevel_(3003), "%s.%s = %llu  (avg=%.2lf  min=%llu  max=%llu  count=%llu)", name.c_str(), EquipmentStatsNames[i], (unsigned long long)equipmentStats[i].get(), equipmentStats[i].getAverage(), (unsigned long long)equipmentStats[i].getMinimum(), (unsigned long long)equipmentStats[i].getMaximum(), (unsigned long long)equipmentStats[i].getCount());
@@ -253,7 +271,6 @@ Thread::CallbackResult ReadoutEquipment::threadCallback(void* arg)
   if (ptr->usingSoftwareClock) {
     if (ptr->timeframeClock.isTimeout()) {
       ptr->currentTimeframe++;
-      ptr->statsNumberOfTimeframes++;
       ptr->timeframeClock.increment();
     }
   }
@@ -301,38 +318,63 @@ Thread::CallbackResult ReadoutEquipment::threadCallback(void* arg)
 
       // get next block
       DataBlockContainerReference nextBlock = nullptr;
-      try {
-        nextBlock = ptr->getNextBlock();
-      } catch (...) {
-        theLog.log(LogWarningDevel_(3230), "getNextBlock() exception");
-        break;
-      }
-      // printf("getNextBlock=%p\n",nextBlock);
-      if (nextBlock == nullptr) {
-        break;
-      }
-
-      // handle RDH-formatted data
-      if (ptr->isRdhEquipment) {
-        ptr->processRdh(nextBlock);
-      }
+      if (ptr->throttlePendingBlock != nullptr) {      
+        nextBlock = std::move(ptr->throttlePendingBlock);
+      } else {
+	try {
+          nextBlock = ptr->getNextBlock();
+	} catch (...) {
+          theLog.log(LogWarningDevel_(3230), "getNextBlock() exception");
+          break;
+	}
 
-      // tag data with equipment Id, if set (will overwrite field if was already set by equipment)
-      if (ptr->id != undefinedEquipmentId) {
-        nextBlock->getData()->header.equipmentId = ptr->id;
+	if (nextBlock == nullptr) {
+          break;
+	}
+
+	// handle RDH-formatted data
+	if (ptr->isRdhEquipment) {
+          ptr->processRdh(nextBlock);
+	}
+	
+	// tag data with equipment Id, if set (will overwrite field if was already set by equipment)
+	if (ptr->id != undefinedEquipmentId) {
+          nextBlock->getData()->header.equipmentId = ptr->id;
+	}
+
+	// tag data with block id
+	ptr->currentBlockId++; // don't start from 0
+	nextBlock->getData()->header.blockId = ptr->currentBlockId;
+
+	// tag data with (dummy) timeframeid, if none set
+	if (nextBlock->getData()->header.timeframeId == undefinedTimeframeId) {
+          nextBlock->getData()->header.timeframeId = ptr->getCurrentTimeframe();
+	}
+
+	// tag data with run number
+	nextBlock->getData()->header.runNumber = occRunNumber;
       }
-
-      // tag data with block id
-      ptr->currentBlockId++; // don't start from 0
-      nextBlock->getData()->header.blockId = ptr->currentBlockId;
-
-      // tag data with (dummy) timeframeid, if none set
-      if (nextBlock->getData()->header.timeframeId == undefinedTimeframeId) {
-        nextBlock->getData()->header.timeframeId = ptr->getCurrentTimeframe();
+      
+      // check TF id of new block
+      uint64_t tfId = nextBlock->getData()->header.timeframeId;
+      if (tfId != ptr->lastTimeframe) {
+	// regulate TF rate if needed
+	if (!ptr->TFregulator.next()) {
+          ptr->throttlePendingBlock = std::move(nextBlock); // keep block with new TF for later
+	  isActive = false; // ask for delay before retry
+	  break;
+	}
+
+	ptr->statsNumberOfTimeframes++;
+
+	// detect gaps in TF id continuity
+	if (tfId != ptr->lastTimeframe + 1) {
+	  if (ptr->cfgRdhDumpWarningEnabled) {
+            theLog.log(LogWarningSupport_(3004), "Non-contiguous timeframe IDs %llu ... %llu", (unsigned long long)ptr->lastTimeframe, (unsigned long long)tfId);
+	  }
+	}
       }
-
-      // tag data with run number
-      nextBlock->getData()->header.runNumber = occRunNumber;
+      ptr->lastTimeframe = tfId;
 
       // update rate-limit clock
       if (ptr->readoutRate > 0) {
@@ -433,13 +475,15 @@ void ReadoutEquipment::initCounters()
   statsNumberOfTimeframes = 0;
 
   // reset timeframe clock
+  currentTimeframe = undefinedTimeframeId;
+  lastTimeframe = undefinedTimeframeId;
+  firstTimeframeHbOrbitBegin = undefinedOrbit;
+  isDefinedFirstTimeframeHbOrbitBegin = 0;
   if (usingSoftwareClock) {
     timeframeClock.reset(1000000 / timeframeRate);
+    currentTimeframe = 1;
   }
 
-  currentTimeframe = undefinedTimeframeId;
-  firstTimeframeHbOrbitBegin = undefinedOrbit;
-  isDefinedFirstTimeframeHbOrbitBegin = 0;
 };
 
 void ReadoutEquipment::finalCounters()
@@ -449,27 +493,13 @@ void ReadoutEquipment::finalCounters()
   }
 };
 
-void ReadoutEquipment::initRdhEquipment() { isRdhEquipment = true; }
-
 uint64_t ReadoutEquipment::getTimeframeFromOrbit(uint32_t hbOrbit)
 {
   if (!isDefinedFirstTimeframeHbOrbitBegin) {
     firstTimeframeHbOrbitBegin = hbOrbit;
     isDefinedFirstTimeframeHbOrbitBegin = 1;
   }
   uint64_t tfId = 1 + (hbOrbit - firstTimeframeHbOrbitBegin) / getTimeframePeriodOrbits();
-  if (tfId != currentTimeframe) {
-    // theLog.log(LogDebugTrace, "TF %lu", tfId);
-    statsNumberOfTimeframes++;
-
-    // detect gaps in TF id continuity
-    if (tfId != currentTimeframe + 1) {
-      if (cfgRdhDumpWarningEnabled) {
-        theLog.log(LogWarningSupport_(3004), "Non-contiguous timeframe IDs %llu ... %llu", (unsigned long long)currentTimeframe, (unsigned long long)tfId);
-      }
-    }
-  }
-  currentTimeframe = tfId;
   return tfId;
 }
 

src/ReadoutEquipment.h

@@ -21,13 +21,16 @@
 #include "MemoryBankManager.h"
 #include "MemoryHandler.h"
 #include "RdhUtils.h"
+#include "RateRegulator.h"
 
 using namespace AliceO2::Common;
 
 class ReadoutEquipment
 {
  public:
-  ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint);
+  // constructor parameters:
+  // - isRdhEquipment: to be set by equipments producing RDH-formatted data. Done here so that appropriate defaults can be used in constructor.
+  ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, bool isRdhEquipment = 0);
   virtual ~ReadoutEquipment();
 
   DataBlockContainerReference getBlock();
@@ -136,7 +139,8 @@ class ReadoutEquipment
   AliceO2::Common::Timer timeframeClock; // timeframe id should be increased at each clock cycle
   uint64_t currentTimeframe = 0;         // id of current timeframe
   bool usingSoftwareClock = false;       // if set, using internal software clock to generate timeframe id
-
+  uint64_t lastTimeframe = 0;            // id of last timeframe (in data)
+  
   //const unsigned int LHCBunches = 3564;    // number of bunches in LHC
   const unsigned int LHCOrbitRate = 11246; // LHC orbit rate, in Hz. 299792458 / 26659
   uint32_t timeframePeriodOrbits = 256;    // timeframe interval duration in number of LHC orbits
@@ -150,6 +154,9 @@ class ReadoutEquipment
   int cfgRdhDumpWarningEnabled = 0;    // flag to enable RDH warning log at runtime
   int cfgRdhUseFirstInPageEnabled = 0; // flag to enable reading of first RDH in page to populate readout headers
   //int cfgRdhCheckPacketCounterContiguous = 1; // flag to enable checking if RDH packetCounter value contiguous (done link-by-link)
+  double cfgTfRateLimit = 0;           // TF rate limit, to throttle data readout
+  RateRegulator TFregulator;           // clock counter for TF rate checks
+  DataBlockContainerReference throttlePendingBlock; // in case TF rate limit was reached, a block may be set aside for later (when it belongs to next TF)
 
   bool isRdhEquipment = false; // to be set true for RDH equipments
 
@@ -166,8 +173,6 @@ class ReadoutEquipment
   // compute range of orbits for given timeframe
   void getTimeframeOrbitRange(uint64_t tfId, uint32_t& hbOrbitMin, uint32_t& hbOrbitMax);
 
-  void initRdhEquipment(); // to be called by equipments producing RDH-formatted data
-
   unsigned long long statsRdhCheckOk = 0;        // number of RDH structs which have passed check ok
   unsigned long long statsRdhCheckErr = 0;       // number of RDH structs which have not passed check
   unsigned long long statsRdhCheckStreamErr = 0; // number of inconsistencies in RDH stream (e.g. ids/timing compared to previous RDH)