Merge pull request #244 from sy-c/master

v2.17.1

Author: sy-c

CMakeLists.txt

@@ -69,7 +69,7 @@ find_package(Configuration)
 find_package(Numa)
 find_package(FairMQ)
 find_package(ZLIB)
-find_package(RDMA)
+#find_package(RDMA)
 find_package(Occ)
 find_package(BookkeepingApi)
 find_package(ZMQ)

doc/configurationParameters.md

@@ -103,6 +103,7 @@ The parameters related to 3rd-party libraries are described here for convenience
 | equipment-* | dataPagesLogPath | string |  | Path where to save a summary of each data pages generated by equipment. |
 | equipment-* | debugFirstPages | int | 0 | If set, print debug information for first (given number of) data pages readout. |
 | equipment-* | disableOutput | int | 0 | If non-zero, data generated by this equipment is discarded immediately and is not pushed to output fifo of readout thread. Used for testing. |
+| equipment-* | dropPagesWithError | int | 0 | If set, the pages with RDH errors are discarded (requires rdhCheckEnabled or rdhUseFirstInPage). |
 | equipment-* | enabled | int | 1 | Enable (value=1) or disable (value=0) the equipment. |
 | equipment-* | equipmentType | string |  | The type of equipment to be instanciated. One of: dummy, rorc, cruEmulator |
 | equipment-* | firstPageOffset | bytes | | Offset of the first page, in bytes from the beginning of the memory pool. If not set (recommended), will start at memoryPoolPageSize (one free page is kept before the first usable page for readout internal use). |

doc/releaseNotes.md

@@ -524,3 +524,9 @@ This file describes the main feature changes for each readout.exe released versi
 ## v2.17.0 - 01/03/2022
 - Updated configuration syntax: section names ending with `-*` can be used to define default parameters. They are applied to all section with similar names. Existing key-value pairs are not overwritten, but are defined according to defaults if they don't exist. For example, it is possible to define the TFperiod for all equipments by adding a section named `[equipment-*]` with `TFperiod=32`.
 - Updated readout to new bookkeeping API.
+
+## v2.17.1 - 08/03/2022
+- Updated configuration parameters:
+  - added equipment-*.dropPagesWithError: if set, the pages with RDH errors are discarded (requires rdhCheckEnabled or rdhUseFirstInPage). This may be used if downstream software is not robust to RDH errors.
+- Disabled unused RDMA features (still available by switch in CMake)
+- Added feature for memory banks real time monitoring. Enabled by setting membanksMonitorRate in /etc/o2.d/readout-defaults.cfg. Output (high-rate text) can be seen locally in real time with `tail -f /tmp/readout-monitor-mempool-(id)`. The status of each page in the bank is displayed.

src/MemoryBankManager.cxx

@@ -14,12 +14,19 @@
 #include <unistd.h>
 #include <sys/mman.h>
 #include <thread>
+#include <Common/Timer.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+
 
 #include "readoutInfoLogger.h"
 
-MemoryBankManager::MemoryBankManager() {}
+MemoryBankManager::MemoryBankManager() {
+}
 
-MemoryBankManager::~MemoryBankManager() {}
+MemoryBankManager::~MemoryBankManager() {
+  stopMonitoring();
+}
 
 int MemoryBankManager::addBank(std::shared_ptr<MemoryBank> bankPtr, std::string name)
 {
@@ -39,6 +46,15 @@ int MemoryBankManager::addBank(std::shared_ptr<MemoryBank> bankPtr, std::string
   return 0;
 }
 
+std::string MemoryBankManager::getMonitorFifoPath(int id) {
+  if (id < 0) {
+    return monitorPath;
+  }
+  char fn[128];
+  snprintf(fn,sizeof(fn),"%s-%d", monitorPath.c_str(), id);
+  return fn;
+}
+
 std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize, size_t pageNumber, std::string bankName, size_t firstPageOffset, size_t blockAlign, int numaNode)
 {
 
@@ -191,6 +207,13 @@ std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize
   std::shared_ptr<MemoryPagesPool> mpp;
   try {
     mpp = std::make_shared<MemoryPagesPool>(pageSize, pageNumber, &(((char*)baseAddress)[offset]), blockSize, nullptr, firstPageOffset, newId);
+    if (mpp != nullptr) {
+      // create FIFO for monitoring
+      mkfifo(getMonitorFifoPath(newId).c_str(), S_IRUSR | S_IWUSR | S_IRGRP| S_IROTH);
+      // keep reference to created pool for monitoring purpose
+      std::unique_lock<std::mutex> lock(bankMutex);
+      pools.push_back(mpp);
+    }
   }
   catch (int err) {
     theLog.log(LogErrorSupport_(3230), "Can not create memory pool from bank: error %d", err);
@@ -220,11 +243,64 @@ int MemoryBankManager::getMemoryRegions(std::vector<memoryRange>& ranges)
 void MemoryBankManager::reset()
 {
   std::unique_lock<std::mutex> lock(bankMutex);
+  // release references to page pools
+  pools.clear();
+  // release banks
   for (auto& it : banks) {
     int useCount = it.bank.use_count();
     theLog.log(LogInfoDevel_(3008), "Releasing bank %s%s", it.name.c_str(), (useCount == 1) ? "" : "warning - still in use elsewhere !");
   }
   banks.clear();
   poolIndex = -1;
+  stopMonitoring();
 }
 
+void MemoryBankManager::monitorThLoop() {
+  AliceO2::Common::Timer t;
+  t.reset(1000000.0 / monitorUpdateRate);
+  for(;!monitorThShutdown.load();) {
+    if (t.isTimeout()) {
+      std::unique_lock<std::mutex> lock(bankMutex);
+      for (auto& it : pools) {
+        // it->getId()
+        //printf("%s\n", it->getDetailedStats().c_str());
+        FILE *fp=fopen(getMonitorFifoPath(it->getId()).c_str(),"w+");
+        if (fp!=NULL) {
+        //\e[3J
+          fprintf(fp,"\ec%s\n\n", it->getDetailedStats().c_str());
+          fclose(fp);
+        }
+      }
+      t.increment();
+    } else {
+      std::this_thread::sleep_for(std::chrono::microseconds(10000));
+    }
+  }
+}
+
+
+void MemoryBankManager::startMonitoring(double v_updateRate, const char* v_monitorPath) {
+  if (monitorTh != nullptr) {
+    stopMonitoring();
+  }
+  if (v_updateRate <= 0) {
+    return;
+  }
+  monitorUpdateRate = v_updateRate;
+  if ((v_monitorPath != nullptr) && (strlen(v_monitorPath) != 0)) {
+    monitorPath = v_monitorPath;
+  } else {
+    monitorPath = monitorPathDefault;
+  }
+  std::function<void(void)> f = std::bind(&MemoryBankManager::monitorThLoop, this);
+  monitorThShutdown = 0;
+  monitorTh=std::make_unique<std::thread>(f);
+}
+
+void MemoryBankManager::stopMonitoring() {
+  if (monitorTh != nullptr) {
+    monitorThShutdown = 1;
+    monitorTh->join();
+    monitorTh = nullptr;
+  }
+}

src/MemoryBankManager.h

@@ -22,6 +22,7 @@
 #include <map>
 #include <memory>
 #include <mutex>
+#include <thread>
 
 #include "MemoryBank.h"
 #include "MemoryPagesPool.h"
@@ -66,10 +67,22 @@ class MemoryBankManager
   // reset bank manager in fresh state, in particular: clear all banks
   void reset();
 
+  void startMonitoring(double updateRate, const char* monitorPath = ""); // starts monitoring at given updateRate and output path
+  void stopMonitoring(); // stop monitoring
+  std::string getMonitorFifoPath(int id); // return full path to monitor pipe for given pool index
+
  private:
   std::vector<bankDescriptor> banks; // list of registered memory banks
   std::mutex bankMutex;              // instance mutex to handle concurrent access to public methods
   int poolIndex = -1;                // an increasing index used to assign a unique id to memory pools
+
+  std::unique_ptr<std::thread> monitorTh; // monitor thread
+  std::atomic<int> monitorThShutdown; // flag to terminate monitor thread
+  void monitorThLoop(); // function launched in monitor thread
+  std::vector<std::shared_ptr<MemoryPagesPool>> pools; // reference to existing page pools, for monitoring purpose
+  double monitorUpdateRate = 0; // monitoring period, in Hz
+  const char *monitorPathDefault = "/tmp/readout-monitor-mempool";
+  std::string monitorPath = monitorPathDefault; // prefix path to output monitoring pipes. "-%d" (bank number) added to it and fifo created.
 };
 
 // a global MemoryBankManager instance

src/MemoryPagesPool.cxx

@@ -545,3 +545,51 @@ void MemoryPage::reportPageStates() {
   }
   printf("\n");
 }
+
+std::string MemoryPagesPool::getDetailedStats() {
+  std::string report;
+  int count[MemoryPage::PageState::Undefined] = {};
+  report += "#";
+  report += std::to_string(getId());
+  report += " ";
+
+  for(const auto &p : pages) {
+    count[p.currentPageState]++;
+  }
+  for (int i=0; i<MemoryPage::PageState::Undefined; i++) {
+    char frac[12];
+    snprintf(frac, sizeof(frac), "%.1f", count[i] * 100.0 / pages.size());
+    report += MemoryPage::getPageStateString((MemoryPage::PageState)i);
+    //report += std::to_string(i);
+    report += ": ";
+    report += frac;
+    report += " %   ";
+  }
+  report+='\n';
+  for(const auto &p : pages) {
+    char s='o';
+    switch(p.currentPageState) {
+      case MemoryPage::Idle:
+        s='.'; break;
+      case MemoryPage::Allocated:
+        s='a'; break;
+      case MemoryPage::InROC:
+        s='C'; break;
+      case MemoryPage::InAggregator:
+        s='A'; break;
+      case MemoryPage::InFMQ:
+        s='F'; break;
+      default:
+        break;
+    }
+    report += s;
+  }
+  report+='\n';
+
+  return report;
+}
+
+// todo:
+// add FMQ release rate
+// start/stop/start -> reorder pages in pool FIFO
+// or even on-the-fly, if there i ?

src/MemoryPagesPool.h

@@ -129,6 +129,7 @@ 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
+  std::string getDetailedStats(); // return detailed stats
 
   // an optional user-provided logging function for all memory pool related ops (including warnings on low)
   typedef std::function<void(const std::string &)> LogCallback;

src/ReadoutEquipment.cxx

@@ -145,7 +145,9 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
   cfg.getOptionalValue<int>(cfgEntryPoint + ".rdhCheckDetectorField", cfgRdhCheckDetectorField);  
   // configuration parameter: | equipment-* | rdhCheckTrigger | int | 0 | If set, the RDH trigger counters are checked for consistency. |
   cfg.getOptionalValue<int>(cfgEntryPoint + ".rdhCheckTrigger", cfgRdhCheckTrigger);
-  theLog.log(LogInfoDevel_(3002), "RDH settings: rdhCheckEnabled=%d rdhDumpEnabled=%d rdhDumpErrorEnabled=%d rdhDumpWarningEnabled=%d rdhUseFirstInPageEnabled=%d rdhCheckFirstOrbit=%d rdhCheckDetectorField=%d", cfgRdhCheckEnabled, cfgRdhDumpEnabled, cfgRdhDumpErrorEnabled, cfgRdhDumpWarningEnabled, cfgRdhUseFirstInPageEnabled, cfgRdhCheckFirstOrbit, cfgRdhCheckDetectorField);
+  // configuration parameter: | equipment-* | dropPagesWithError | int | 0 | If set, the pages with RDH errors are discarded (requires rdhCheckEnabled or rdhUseFirstInPage). |
+  cfg.getOptionalValue<int>(cfgEntryPoint + ".dropPagesWithError", cfgDropPagesWithError);
+  theLog.log(LogInfoDevel_(3002), "RDH settings: rdhCheckEnabled=%d rdhDumpEnabled=%d rdhDumpErrorEnabled=%d rdhDumpWarningEnabled=%d rdhUseFirstInPageEnabled=%d rdhCheckFirstOrbit=%d rdhCheckDetectorField=%d dropPagesWithError=%d", cfgRdhCheckEnabled, cfgRdhDumpEnabled, cfgRdhDumpErrorEnabled, cfgRdhDumpWarningEnabled, cfgRdhUseFirstInPageEnabled, cfgRdhCheckFirstOrbit, cfgRdhCheckDetectorField, cfgDropPagesWithError);
 
   // configuration parameter: | equipment-* | ctpMode | int | 0 | If set, the detector field (CTP run mask) is checked. Incoming data is discarded until a new bit is set, and discarded again after this bit is unset. Automatically implies rdhCheckDetectorField=1 and rdhCheckDetectorField=1. |
   cfg.getOptionalValue<int>(cfgEntryPoint + ".ctpMode", cfgCtpMode);
@@ -336,7 +338,7 @@ ReadoutEquipment::~ReadoutEquipment()
 
   // 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());
+    theLog.log(LogInfoDevel_(3008), "Equipment %s :  mempool still has %d references", name.c_str(), (int)mp.use_count());
   }
 
   if (fpDataPagesLog != nullptr) {
@@ -430,7 +432,14 @@ Thread::CallbackResult ReadoutEquipment::threadCallback(void* arg)
 
 	// handle RDH-formatted data
 	if (ptr->cfgRdhUseFirstInPageEnabled) {
-          ptr->processRdh(nextBlock);
+          if ((ptr->processRdh(nextBlock)) && ptr->cfgDropPagesWithError) {
+            // drop pages with error when configured to do so
+            ptr->statsRdhCheckPagesDropped++;
+            static InfoLogger::AutoMuteToken logPageErrorDrop(LogWarningSupport_(3235), 10, 60);
+            theLog.log(logPageErrorDrop, "Equipment %s : page with RDH error has been discarded (total: %llu)", ptr->name.c_str(), ptr->statsRdhCheckPagesDropped);
+            nextBlock = nullptr;
+            continue;
+          }
 	}
 
         // discard data immediately if configured to do so
@@ -593,6 +602,7 @@ void ReadoutEquipment::initCounters()
   statsRdhCheckOk = 0;
   statsRdhCheckErr = 0;
   statsRdhCheckStreamErr = 0;
+  statsRdhCheckPagesDropped = 0;
 
   statsNumberOfTimeframes = 0;
 
@@ -621,7 +631,8 @@ void ReadoutEquipment::initCounters()
 void ReadoutEquipment::finalCounters()
 {
   if (cfgRdhCheckEnabled) {
-    theLog.log(LogInfoDevel_(3003), "Equipment %s : %llu timeframes, RDH checks %llu ok, %llu errors, %llu stream inconsistencies", name.c_str(), statsNumberOfTimeframes, statsRdhCheckOk, statsRdhCheckErr, statsRdhCheckStreamErr);
+    theLog.log(LogInfoDevel_(3003), "Equipment %s : %llu timeframes, RDH checks %llu ok, %llu errors, %llu stream inconsistencies, %llu pages with error dropped",
+      name.c_str(), statsNumberOfTimeframes, statsRdhCheckOk, statsRdhCheckErr, statsRdhCheckStreamErr, statsRdhCheckPagesDropped);
   }
 };
 
@@ -1037,7 +1048,7 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
     }
   }
 
-  return 0;
+  return isPageError;
 }
 
 void ReadoutEquipment::abortThread() {

src/ReadoutEquipment.h

@@ -203,6 +203,7 @@ class ReadoutEquipment
   int saveErrorPagesCount; // counter for number of pages dumped so far
   std::string cfgDataPagesLogPath; // path to write to disk a summary for each data page received
   FILE *fpDataPagesLog = nullptr; // handle to data page log file
+  int cfgDropPagesWithError = 0; // if set, pages with RDH errors are dropped by readout
 
  protected:
   // get timeframe from orbit
@@ -218,6 +219,7 @@ class ReadoutEquipment
   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)
+  unsigned long long statsRdhCheckPagesDropped = 0; // number of pages discarded because of RDH errors (when configured to do so)
 };
 
 std::unique_ptr<ReadoutEquipment> getReadoutEquipmentDummy(ConfigFile& cfg, std::string cfgEntryPoint);

src/ReadoutVersion.h

@@ -9,5 +9,5 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
-#define READOUT_VERSION "2.17.0"
+#define READOUT_VERSION "2.17.1"