Merge pull request #193 from sy-c/master

features

Author: sy-c

doc/releaseNotes.md

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

src/ConsumerFMQchannel.cxx

@@ -188,34 +188,44 @@ class ConsumerFMQchannel : public Consumer
 
       // 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 {
 
-	}
+      // configuration parameter: | consumer-FairMQChannel-* | checkResources | string | | Check beforehand if unmanaged region would fit in given list of resources. Comma-separated list of items to be checked: eg /dev/shm, MemFree, MemAvailable. (any filesystem path, and any /proc/meminfo entry).|
+      std::string cfgCheckResources;
+      cfg.getOptionalValue<std::string>(cfgEntryPoint + ".checkResources", cfgCheckResources);
+      bool isResourceError = 0;
+      std::vector<std::string> resources;
+      
+      if (getListFromString(cfgCheckResources, resources)) {
+        throw("Can not parse configuration item checkResources");
       }
-            
-      // 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");
+      
+      for(auto r : resources) {
+        unsigned long long freeBytes;
+
+	int getStatsErr = 0;
+        if (r.find_first_of("/")!=std::string::npos) {
+	  // this is a path
+	  getStatsErr = getStatsFilesystem(freeBytes, r);
 	} else {
-          theLog.log(LogInfoSupport_(3230), "Stats shared memory available: %lld MB", (long long)(freeBytes/1048576LL));
+	  // look in /proc/meminfo
+	  getStatsErr = getStatsMemory(freeBytes, r);
+	  r = "/proc/meminfo " + r;
+	}
+	
+	if (getStatsErr) {
+            theLog.log(LogWarningSupport_(3230), "Can not get stats for %s", r.c_str());
+	} else {
+          theLog.log(LogInfoSupport_(3230), "Stats for %s : %lld MB available", r.c_str(), (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(LogErrorSupport_(3230), "Not enough space on %s", r.c_str());
+            isResourceError = 1;
 	  }
 	}
       }
+
+      if (isResourceError) {
+        throw "ConsumerFMQ: can not allocate shared memory region, system resources check failed";
+      }      
 
       theLog.log(LogInfoDevel_(3008), "Creating FMQ unmanaged memory region");
       memoryBuffer = sendingChannel->Transport()->CreateUnmanagedRegion(mMemorySize, [](void* /*data*/, size_t /*size*/, void* hint) { // cleanup callback

src/ReadoutEquipment.cxx

@@ -100,6 +100,10 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
   cfgTfRateLimit = 0;
   cfg.getOptionalValue<double>("readout.tfRateLimit", cfgTfRateLimit);
 
+  // get TF disable flag from toplevel config
+  cfgDisableTimeframes = 0;
+  cfg.getOptionalValue<int>("readout.disableTimeframes", cfgDisableTimeframes);
+
   // 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);
@@ -120,22 +124,24 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
   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);
 
-  // configuration parameter: | equipment-* | TFperiod | int | 256 | Duration of a timeframe, in number of LHC orbits. |
-  int cfgTFperiod = 256;
-  cfg.getOptionalValue<int>(cfgEntryPoint + ".TFperiod", cfgTFperiod);
-  timeframePeriodOrbits = cfgTFperiod;
+  if (!cfgDisableTimeframes) {
+    // configuration parameter: | equipment-* | TFperiod | int | 256 | Duration of a timeframe, in number of LHC orbits. |
+    int cfgTFperiod = 256;
+    cfg.getOptionalValue<int>(cfgEntryPoint + ".TFperiod", cfgTFperiod);
+    timeframePeriodOrbits = cfgTFperiod;
 
-  if (!cfgRdhUseFirstInPageEnabled) {
-    usingSoftwareClock = true; // if RDH disabled, use internal clock for TF id
-  }
-  theLog.log(LogInfoDevel_(3002), "Timeframe length = %d orbits", (int)timeframePeriodOrbits);
-  if (usingSoftwareClock) {
-    timeframeRate = LHCOrbitRate * 1.0 / timeframePeriodOrbits; // timeframe rate, in Hz
-    theLog.log(LogInfoDevel_(3002), "Timeframe IDs generated by software, %.2lf Hz", timeframeRate);
-  } else {
-    theLog.log(LogInfoDevel_(3002), "Timeframe IDs generated from RDH trigger counters");
+    if (!cfgRdhUseFirstInPageEnabled) {
+      usingSoftwareClock = true; // if RDH disabled, use internal clock for TF id
+    }
+    theLog.log(LogInfoDevel_(3002), "Timeframe length = %d orbits", (int)timeframePeriodOrbits);
+    if (usingSoftwareClock) {
+      timeframeRate = LHCOrbitRate * 1.0 / timeframePeriodOrbits; // timeframe rate, in Hz
+      theLog.log(LogInfoDevel_(3002), "Timeframe IDs generated by software, %.2lf Hz", timeframeRate);
+    } else {
+      theLog.log(LogInfoDevel_(3002), "Timeframe IDs generated from RDH trigger counters");
+    }
   }
-
+  
   // init stats
   equipmentStats.resize(EquipmentStatsIndexes::maxIndex);
   equipmentStatsLast.resize(EquipmentStatsIndexes::maxIndex);
@@ -346,9 +352,14 @@ Thread::CallbackResult ReadoutEquipment::threadCallback(void* arg)
 	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();
+	if (ptr->cfgDisableTimeframes) {
+	  // disable TF id
+          nextBlock->getData()->header.timeframeId = undefinedTimeframeId;
+	} else {
+	  // 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
@@ -647,14 +658,16 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
       }
 
       // check no timeframe overlap in page
-      if (((blockHeader.timeframeOrbitFirst < blockHeader.timeframeOrbitLast) && ((h.getTriggerOrbit() < blockHeader.timeframeOrbitFirst) || (h.getTriggerOrbit() > blockHeader.timeframeOrbitLast))) || ((blockHeader.timeframeOrbitFirst > blockHeader.timeframeOrbitLast) && ((h.getTriggerOrbit() < blockHeader.timeframeOrbitFirst) && (h.getTriggerOrbit() > blockHeader.timeframeOrbitLast)))) {
-        if (cfgRdhDumpErrorEnabled) {
-          theLog.log(logRdhErrorsToken, "Equipment %d RDH #%d @ 0x%X : TimeFrame ID change in page not allowed : orbit 0x%08X not in range [0x%08X,0x%08X]", id, rdhIndexInPage, (unsigned int)pageOffset, (int)h.getTriggerOrbit(), (int)blockHeader.timeframeOrbitFirst, (int)blockHeader.timeframeOrbitLast);
-        }
-        statsRdhCheckStreamErr++;
-        break; // stop checking this page
+      if (!cfgDisableTimeframes) {
+	if (((blockHeader.timeframeOrbitFirst < blockHeader.timeframeOrbitLast) && ((h.getTriggerOrbit() < blockHeader.timeframeOrbitFirst) || (h.getTriggerOrbit() > blockHeader.timeframeOrbitLast))) || ((blockHeader.timeframeOrbitFirst > blockHeader.timeframeOrbitLast) && ((h.getTriggerOrbit() < blockHeader.timeframeOrbitFirst) && (h.getTriggerOrbit() > blockHeader.timeframeOrbitLast)))) {
+          if (cfgRdhDumpErrorEnabled) {
+            theLog.log(logRdhErrorsToken, "Equipment %d RDH #%d @ 0x%X : TimeFrame ID change in page not allowed : orbit 0x%08X not in range [0x%08X,0x%08X]", id, rdhIndexInPage, (unsigned int)pageOffset, (int)h.getTriggerOrbit(), (int)blockHeader.timeframeOrbitFirst, (int)blockHeader.timeframeOrbitLast);
+          }
+          statsRdhCheckStreamErr++;
+          break; // stop checking this page
+	}
       }
-
+      
       /*
       // check packetCounter is contiguous
       if (cfgRdhCheckPacketCounterContiguous) {

src/ReadoutEquipment.h

@@ -155,6 +155,7 @@ class ReadoutEquipment
   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
+  int cfgDisableTimeframes = 0;        // When set, all TF features disabled
   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)
 

src/ReadoutUtils.cxx

@@ -181,18 +181,48 @@ int getIntegerListFromString(const std::string& input, std::vector<int>& output)
   return 0;
 }
 
-int getStatsFreeMemory(unsigned long long &freeBytes) {
+
+// check if a string made of only of simple chars
+// arbitrary selection: letters, digits, ()_
+bool isSimpleString(const std::string &str) {
+     return find_if_not(str.begin(), str.end(), 
+        [](char c) { return (isalnum(c) || (c == '(') || (c == ')')|| (c == '_')); }) == str.end();
+}
+
+
+int getListFromString(const std::string& input, std::vector<std::string>& output)
+{
+  // coma-separated list of simple strings
+  std::istringstream f(input);
+  std::string s;
+  while (std::getline(f, s, ',')) {
+    // trim
+    const std::string blanks = "\t\n\v\f\r ";
+    s.erase(s.find_last_not_of(blanks) + 1);
+    s.erase(0, s.find_first_not_of(blanks));
+    output.push_back(s);
+  }
+  return 0;
+}
+
+int getStatsMemory(unsigned long long &freeBytes, const std::string &keyword) {
   FILE *fp;
   const int lineBufSz = 128;
   char lineBuf[lineBufSz];
   long long value;
   int success = 0;
   freeBytes = 0;
 
+  // check if keyword looks suspicious
+  if (!isSimpleString(keyword)) {
+    return -1;
+  }
+    
   if ((fp = fopen("/proc/meminfo", "r")) != NULL) {
 
+    std::string entryLine = keyword + ": %lld kB";
     while (fgets(lineBuf, lineBufSz, fp) != NULL) {
-      if ( sscanf(lineBuf, "MemFree: %lld kB", &value) == 1) {
+      if ( sscanf(lineBuf, entryLine.c_str(), &value) == 1) {
 	freeBytes = ((unsigned long long)value) * 1024;
 	success = 1;
 	break;
@@ -208,12 +238,12 @@ int getStatsFreeMemory(unsigned long long &freeBytes) {
   return 0;
 }
 
-int getStatsFreeSHM(unsigned long long &freeBytes) {
+int getStatsFilesystem(unsigned long long &freeBytes, const std::string &path) {
   int success = 0;
   freeBytes = 0;
 
   try {
-    freeBytes = (unsigned long long) (std::filesystem::space("/dev/shm")).free;
+    freeBytes = (unsigned long long) (std::filesystem::space(path)).free;
     success = 1;
   }
   catch (...) {

src/ReadoutUtils.h

@@ -45,6 +45,10 @@ int getKeyValuePairsFromString(const std::string& input, std::map<std::string, s
 // returns 0 on success, -1 on error
 int getIntegerListFromString(const std::string& input, std::vector<int>& output);
 
+// parse a string of coma-separated strings into a vector
+// returns 0 on success, -1 on error
+int getListFromString(const std::string& input, std::vector<std::string>& output);
+
 // function to convert a value in bytes to a prefixed number 3+3 digits
 // suffix is the "base unit" to add after calculated prefix, e.g. Byte-> kBytes
 std::string NumberOfBytesToString(double value, const char* suffix, int base = 1024);
@@ -56,15 +60,15 @@ int getProcessStats(double& uTime, double& sTime);
 typedef uint32_t tRunNumber;
 typedef uint32_t tTimeframeId;
 
-// function to retrieve amount of free memory on the system
+// function to retrieve some memory statistics on the system
 // Works only when /proc/meminfo available
+// Look for entry corresponding to provided keyword (eg: MemFree, MemAvailable)
 // returns 0 on success, -1 on error
-int getStatsFreeMemory(unsigned long long &freeBytes);
+int getStatsMemory(unsigned long long &freeBytes, const std::string& keyword);
 
-// function to retrieve amount of free memory on the system
-// Works only when /dev/shm available
+// function to retrieve amount of free area for given path on the filesystem
 // returns 0 on success, -1 on error
-int getStatsFreeSHM(unsigned long long &freeBytes);
+int getStatsFilesystem(unsigned long long &freeBytes, const std::string& path);
 
 // end of _READOUTUTILS_H
 #endif

src/mainReadout.cxx

@@ -143,6 +143,7 @@ class Readout
   // configuration parameters
   double cfgExitTimeout;
   double cfgFlushEquipmentTimeout;
+  int cfgDisableTimeframes;
   int cfgDisableAggregatorSlicing;
   double cfgAggregatorSliceTimeout;
   double cfgAggregatorStfTimeout;
@@ -468,10 +469,21 @@ int Readout::configure(const boost::property_tree::ptree& properties)
   // configuration parameter: | readout | tfRateLimit | double | 0 | When set, the output is limited to a given timeframe rate. |
   cfgTfRateLimit = 0;
   cfg.getOptionalValue<double>("readout.tfRateLimit", cfgTfRateLimit);
+
+  // configuration parameter: | readout | disableTimeframes | int | 0 | When set, all timeframe related features are disabled (this may supersede other config parameters). |
+  cfgDisableTimeframes = 0;
+  cfg.getOptionalValue<int>("readout.disableTimeframes", cfgDisableTimeframes);
+  if (cfgDisableTimeframes) {
+    cfgDisableAggregatorSlicing = 1;
+    cfgTfRateLimit = 0;
+    theLog.log(LogInfoDevel, "Timeframes disabled");
+  }
+
   if (cfgTfRateLimit > 0) {
     theLog.log(LogInfoDevel, "Timeframe rate limit = % .2lf Hz", cfgTfRateLimit);
   }
 
+
   // configuration parameter: | readout | logbookEnabled | int | 0 | When set, the logbook is enabled and populated with readout stats at runtime. |
   cfgLogbookEnabled = 0;
   cfg.getOptionalValue<int>("readout.logbookEnabled", cfgLogbookEnabled);
@@ -828,15 +840,16 @@ int Readout::start()
   if (cfgDisableAggregatorSlicing) {
     theLog.log(LogInfoDevel, "Aggregator slicing disabled");
     agg->disableSlicing = 1;
-  }
-  if (cfgAggregatorSliceTimeout > 0) {
-    theLog.log(LogInfoDevel, "Aggregator slice timeout = %.2lf seconds", cfgAggregatorSliceTimeout);
-    agg->cfgSliceTimeout = cfgAggregatorSliceTimeout;
-  }
-  if (cfgAggregatorStfTimeout > 0) {
-    theLog.log(LogInfoDevel, "Aggregator subtimeframe timeout = %.2lf seconds", cfgAggregatorStfTimeout);
-    agg->cfgStfTimeout = cfgAggregatorStfTimeout;
-    agg->enableStfBuilding = 1;
+  } else {
+    if (cfgAggregatorSliceTimeout > 0) {
+      theLog.log(LogInfoDevel, "Aggregator slice timeout = %.2lf seconds", cfgAggregatorSliceTimeout);
+      agg->cfgSliceTimeout = cfgAggregatorSliceTimeout;
+    }
+    if (cfgAggregatorStfTimeout > 0) {
+      theLog.log(LogInfoDevel, "Aggregator subtimeframe timeout = %.2lf seconds", cfgAggregatorStfTimeout);
+      agg->cfgStfTimeout = cfgAggregatorStfTimeout;
+      agg->enableStfBuilding = 1;
+    }
   }
 
   agg->start();