RDH equipments defaults, FMQ stats hidden when not used, tfRateLimit handling, cruemu tf id clock

Author: sy-c

doc/releaseNotes.md

@@ -348,3 +348,7 @@ This file describes the main feature changes for each readout.exe released versi
 
 ## next version
 - 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).

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.|

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)

src/ReadoutEquipmentCruEmulator.cxx

@@ -73,12 +73,9 @@ class ReadoutEquipmentCruEmulator : public ReadoutEquipment
   std::vector<DataBlockContainerReference> pendingBlocks;                          // pages being filled (1 per link)
 };
 
-ReadoutEquipmentCruEmulator::ReadoutEquipmentCruEmulator(ConfigFile& cfg, std::string cfgEntryPoint) : ReadoutEquipment(cfg, cfgEntryPoint)
+ReadoutEquipmentCruEmulator::ReadoutEquipmentCruEmulator(ConfigFile& cfg, std::string cfgEntryPoint) : ReadoutEquipment(cfg, cfgEntryPoint, 1) // this is RDH-data equipment
 {
 
-  // declare RDH equipment
-  initRdhEquipment();
-
   // get configuration values
   // configuration parameter: | equipment-cruemulator-* | maxBlocksPerPage | int | 0 | [obsolete- not used]. Maximum number of blocks per page. |
   // configuration parameter: | equipment-cruemulator-* | cruBlockSize | int | 8192 | Size of a RDH block. |
@@ -182,7 +179,7 @@ Thread::CallbackResult ReadoutEquipmentCruEmulator::prepareBlocks()
 
     nowOrbit = LHCorbit;
     nowBc = LHCbc;
-    uint64_t nowId = getCurrentTimeframe();
+    uint64_t nowId = getTimeframeFromOrbit(nowOrbit);
 
     int linkId = cfgLinkId + currentLink;
     int bytesAvailableInPage = b->header.dataSize; // a bit less than memoryPoolPageSize;
@@ -287,7 +284,6 @@ Thread::CallbackResult ReadoutEquipmentCruEmulator::prepareBlocks()
     // no need to fill header defaults, this is done by getNewDataBlockContainer()
     // only adjust payload size
     b->header.dataSize = dSize;
-    b->header.timeframeId = nowId;
     b->header.linkId = linkId;
 
     readyBlocks->push(pendingBlocks[currentLink]);

src/ReadoutEquipmentPlayer.cxx

@@ -75,12 +75,9 @@ void ReadoutEquipmentPlayer::copyFileDataToPage(void* page)
   }
 }
 
-ReadoutEquipmentPlayer::ReadoutEquipmentPlayer(ConfigFile& cfg, std::string cfgEntryPoint) : ReadoutEquipment(cfg, cfgEntryPoint)
+ReadoutEquipmentPlayer::ReadoutEquipmentPlayer(ConfigFile& cfg, std::string cfgEntryPoint) : ReadoutEquipment(cfg, cfgEntryPoint, 1)  // this is RDH-data equipment
 {
 
-  // declare RDH equipment
-  initRdhEquipment();
-
   auto errorHandler = [&](const std::string& err) {
     if (fp != nullptr) {
       fclose(fp);

src/ReadoutEquipmentRORC.cxx

@@ -70,12 +70,9 @@ class ReadoutEquipmentRORC : public ReadoutEquipment
 struct ReadoutEquipmentRORCException : virtual Exception {
 };
 
-ReadoutEquipmentRORC::ReadoutEquipmentRORC(ConfigFile& cfg, std::string name) : ReadoutEquipment(cfg, name)
+ReadoutEquipmentRORC::ReadoutEquipmentRORC(ConfigFile& cfg, std::string name) : ReadoutEquipment(cfg, name, 1)  // this is RDH-data equipment
 {
 
-  // declare RDH equipment
-  initRdhEquipment();
-
   try {
 
     // get parameters from configuration

src/ReadoutStats.h

@@ -46,6 +46,7 @@ class ReadoutStats
   void print();
 
   ReadoutStatsCounters counters;
+  bool isFairMQ; // flag to report when FairMQ used
 };
 
 extern ReadoutStats gReadoutStats;

src/mainReadout.cxx

@@ -295,6 +295,9 @@ int Readout::configure(const boost::property_tree::ptree& properties)
 {
   theLog.log(LogInfoSupport_(3005), "Readout executing CONFIGURE");
 
+  // reset some flags
+  gReadoutStats.isFairMQ = 0; // disable FMQ stats
+
   // load configuration file
   theLog.log(LogInfoSupport, "Reading configuration from %s %s", cfgFileURI, cfgFileEntryPoint);
   try {