Merge pull request #220 from sy-c/master

v2.10.2

Author: sy-c

doc/releaseNotes.md

@@ -451,3 +451,6 @@ This file describes the main feature changes for each readout.exe released versi
 ## v2.10.1 - 20/04/2022
 - Updated configuration parameters:
   - equipment-file-*.updateOrbits: when set to zero, RDH orbits are not updated in file loop replay. This is needed for some reconstruction tests. This however creates a stream of data with inconsistent orbit ids and mismatching timeframe information.
+
+## v2.10.2 - 04/05/2022
+- Added some buffer monitoring counters. They are logged in console stats and in monitoring (readout.bufferUsage, one per buffer, from 0% to 100%).

src/ConsumerFMQchannel.cxx

@@ -286,8 +286,11 @@ class ConsumerFMQchannel : public Consumer
       throw "ConsumerFMQ: failed to get memory pool from " + memoryBankName + " for " + std::to_string(memoryPoolNumberOfPages) + " pages x " + std::to_string(memoryPoolPageSize) + " bytes";
     } else {
       mp -> setWarningCallback(std::bind(&ConsumerFMQchannel::mplog, this, std::placeholders::_1));
+      if ((mp->getId() >= 0) && (mp->getId() < ReadoutStatsMaxItems)) {
+	mp -> setBufferStateVariable(&gReadoutStats.counters.bufferUsage[mp->getId()]);
+      }
     }
-    theLog.log(LogInfoDevel_(3008), "Using memory pool %d pages x %d bytes", memoryPoolNumberOfPages, memoryPoolPageSize);
+    theLog.log(LogInfoDevel_(3008), "Using memory pool [%d]: %d pages x %d bytes", mp->getId(), memoryPoolNumberOfPages, memoryPoolPageSize);
   }
 
   ~ConsumerFMQchannel()

src/ConsumerStats.cxx

@@ -156,6 +156,14 @@ class ConsumerStats : public Consumer
       sendMetricNoException({ rRfmq, "readout.stfbMemoryPagesReleaseRate"});
       sendMetricNoException({ avgTfmq, "readout.stfbMemoryPagesReleaseLatency"});
       sendMetricNoException({ tfidfmq, "readout.stfbTimeframeId"});
+
+      // buffer stats
+      for (int i = 0; i < ReadoutStatsMaxItems; i++) {
+	double r = snapshot.bufferUsage[i].load();
+	if (r >= 0) {
+	  sendMetricNoException(Metric{(int)(r*100), "readout.bufferUsage"}.addTag(tags::Key::ID, i));
+	}
+      }
     }
 
 #ifdef WITH_ZMQ
@@ -171,6 +179,16 @@ class ConsumerStats : public Consumer
 	if (gReadoutStats.isFairMQ) {
           theLog.log(LogInfoOps_(3003), "STFB locked pages: current=%llu, released = %llu, release rate=%.2lf Hz, latency=%.3lf s, current TF = %d", (unsigned long long) snapshot.pagesPendingFairMQ.load(), nRfmq, rRfmq, avgTfmq, tfidfmq );
 	}
+	std::string bufferReport;
+	for (int i = 0; i < ReadoutStatsMaxItems; i++) {
+	  double r = snapshot.bufferUsage[i].load();
+	  if (r >= 0) {
+	    bufferReport += "["+ std::to_string(i) + "]=" + std::to_string((int)(r*100)) + "% ";
+	  }
+	}
+	if (bufferReport.length()) {
+	  theLog.log(LogInfoOps_(3003), "Memory buffers usage: %s", bufferReport.c_str());
+	}
       }
     }
 

src/MemoryBankManager.cxx

@@ -41,6 +41,7 @@ std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize
   void* baseAddress = nullptr; // base address of bank from which the block is taken
   size_t offset = 0;           // offset of new block (relative to baseAddress)
   size_t blockSize = 0;        // size of new block (in bytes)
+  int newId = 0;
 
   // disable concurrent execution of this block
   // automatic release of lock when going out of scope
@@ -109,11 +110,12 @@ std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize
 
     // keep track of this new block
     banks[ix].rangesInUse.push_back({ offset, blockSize });
+    newId = ++poolIndex;
   }
   // end of locked block
 
   // create pool of pages from new block
-  return std::make_shared<MemoryPagesPool>(pageSize, pageNumber, &(((char*)baseAddress)[offset]), blockSize, nullptr, firstPageOffset);
+  return std::make_shared<MemoryPagesPool>(pageSize, pageNumber, &(((char*)baseAddress)[offset]), blockSize, nullptr, firstPageOffset, newId);
 }
 
 // a global MemoryBankManager instance
@@ -140,5 +142,6 @@ void MemoryBankManager::reset()
     theLog.log(LogInfoDevel_(3008), "Releasing bank %s%s", it.name.c_str(), (useCount == 1) ? "" : "warning - still in use elsewhere !");
   }
   banks.clear();
+  poolIndex = -1;
 }
 

src/MemoryBankManager.h

@@ -68,6 +68,7 @@ class MemoryBankManager
  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
 };
 
 // a global MemoryBankManager instance

src/MemoryPagesPool.cxx

@@ -16,7 +16,7 @@
 
 int MemoryPagesPoolStatsEnabled = 0; // flag to control memory stats
 
-MemoryPagesPool::MemoryPagesPool(size_t vPageSize, size_t vNumberOfPages, void* vBaseAddress, size_t vBaseSize, ReleaseCallback vCallback, size_t firstPageOffset)
+MemoryPagesPool::MemoryPagesPool(size_t vPageSize, size_t vNumberOfPages, void* vBaseAddress, size_t vBaseSize, ReleaseCallback vCallback, size_t firstPageOffset, int vId)
 {
   // initialize members from parameters
   pageSize = vPageSize;
@@ -25,6 +25,7 @@ MemoryPagesPool::MemoryPagesPool(size_t vPageSize, size_t vNumberOfPages, void*
   baseBlockSize = vBaseSize;
   releaseBaseBlockCallback = vCallback;
   state = BufferState::empty;
+  id = vId;
 
   // check page / header sizes
   assert(headerReservedSpace >= sizeof(DataBlock));
@@ -77,6 +78,9 @@ MemoryPagesPool::MemoryPagesPool(size_t vPageSize, size_t vNumberOfPages, void*
     t3.enableHistogram(64, 1, 100000000);
     t4.enableHistogram(64, 1, 100000000);
   }
+
+  // udpate buffer state
+  updateBufferState();
 }
 
 MemoryPagesPool::~MemoryPagesPool()
@@ -192,6 +196,9 @@ void MemoryPagesPool::releasePage(void* address)
     }
   }
 
+  // disable concurrent execution of this function
+  std::unique_lock<std::mutex> lock(pagesAvailableMutexPush);
+
   // put back page in list of available pages
   pagesAvailable->push(address);
 
@@ -302,4 +309,17 @@ void MemoryPagesPool::updateBufferState() {
     state = BufferState::empty;
     log("buffer usage back to reasonable level");
   }
+  if (pBufferState != nullptr) {
+    pBufferState->store(r);
+  }
+}
+
+int MemoryPagesPool::getId() {
+  return id;
+}
+
+void MemoryPagesPool::setBufferStateVariable(std::atomic<double> *bufferStateVar) {
+  pBufferState=bufferStateVar;
+  updateBufferState();
+  printf("buffer usage = %lf @ 0x%p\n", pBufferState->load(),pBufferState);
 }

src/MemoryPagesPool.h

@@ -18,6 +18,7 @@
 #include <map>
 #include <memory>
 #include <string>
+#include <mutex>
 
 #include "CounterStats.h"
 #include "DataBlockContainer.h"
@@ -43,7 +44,8 @@ class MemoryPagesPool
   // - size of memory block in bytes (if zero, assuming it is big enough for page number * page size - not taking into account firstPageOffset is set)
   // - a release callback to be called at destruction time
   // - firstPageOffset is the offset of first page from base address. This is to control alignment. All pages are created contiguous from this point. If non-zero, this may reduce number of pages created compared to request (as to fit in base size)
-  MemoryPagesPool(size_t pageSize, size_t numberOfPages, void* baseAddress, size_t baseSize = 0, ReleaseCallback callback = nullptr, size_t firstPageOffset = 0);
+  // - id: an optional identifier
+  MemoryPagesPool(size_t pageSize, size_t numberOfPages, void* baseAddress, size_t baseSize = 0, ReleaseCallback callback = nullptr, size_t firstPageOffset = 0, int id = -1);
 
   // destructor
   ~MemoryPagesPool();
@@ -59,6 +61,7 @@ class MemoryPagesPool
   size_t getNumberOfPagesAvailable(); // get number of pages currently available
   void* getBaseBlockAddress();        // get the base address of memory pool block
   size_t getBaseBlockSize();          // get the  size of memory pool block. All pages guaranteed to be within &baseBlockAddress[0] and &baseBlockAddress[baseBlockSize]
+  int getId();                        // get pool identifier, as set on creation
 
   // get an empty data block container from the pool
   // parameters:
@@ -77,6 +80,7 @@ class MemoryPagesPool
   typedef std::function<void(const std::string &)> LogCallback;
 
   void setWarningCallback(const LogCallback& cb, double thHigh = 0.9, double thOk = 0.8);
+  void setBufferStateVariable(std::atomic<double> *bufferStateVar); // the provided variable is updated continuously with the buffer usage ratio (0.0 empty -> 1.0 full)
 
  private:
 
@@ -87,8 +91,10 @@ class MemoryPagesPool
   enum BufferState {empty, high, full};
   BufferState state = BufferState::empty;
   void updateBufferState();
+  std::atomic<double> *pBufferState = nullptr; // when set, the pointed variable is updated everytime updateBufferState() is called
 
   std::unique_ptr<AliceO2::Common::Fifo<void*>> pagesAvailable; // a buffer to keep track of individual pages
+  std::mutex pagesAvailableMutexPush; // a lock to avoid concurrent push-back of free pages to fifo
 
   size_t numberOfPages;                           // number of pages
   size_t pageSize;                                // size of each page, in bytes
@@ -124,6 +130,7 @@ class MemoryPagesPool
   CounterStats t1, t2, t3, t4;
 
   CounterStats poolStats; // keep track of number of free pages in the pool 
+  int id = -1; // unique identifier for this pool
 };
 
 #endif // #ifndef _MEMORYPAGESPOOL_H

src/ReadoutEquipment.cxx

@@ -203,6 +203,10 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
     throw __LINE__;
   } else {
     mp -> setWarningCallback(std::bind(&ReadoutEquipment::mplog, this, std::placeholders::_1));
+    if ((mp->getId() >= 0) && (mp->getId() < ReadoutStatsMaxItems)) {
+      mp -> setBufferStateVariable(&gReadoutStats.counters.bufferUsage[mp->getId()]);
+    }
+    theLog.log(LogInfoDevel_(3008), "Using memory pool [%d]: %d pages x %d bytes", mp->getId(), memoryPoolNumberOfPages, memoryPoolPageSize);
   }
   // todo: move page align to MemoryPool class
   assert(pageSpaceReserved == mp->getPageSize() - mp->getDataBlockMaxSize());

src/ReadoutEquipmentRORC.cxx

@@ -210,10 +210,18 @@ Thread::CallbackResult ReadoutEquipmentRORC::prepareBlocks()
       }
     }
     lastPacketDropped = currentPacketDropped;
+
     if (isWaitingFirstLoop) {
       packetDroppedTimer.reset(1000000); // 1 sec interval
     } else {
       packetDroppedTimer.increment();
+      // check CRU FIFO status - but only after first loop, otherwise would be empty yet
+      if (0) { // feature disabled for the time being, spurious warnings on startup
+      if (!channel->areSuperpageFifosHealthy()) {
+	static InfoLogger::AutoMuteToken logToken(LogWarningSupport_(3235), 5, 60);
+	theLog.log(logToken, "Equipment %s: ROC memory fifo not healthy", name.c_str());
+      }
+      }
     }
   }
 

src/ReadoutStats.cxx

@@ -46,7 +46,7 @@ ReadoutStats::~ReadoutStats() {
   zmqCleanup();
 }
 
-void ReadoutStats::reset()
+void ReadoutStats::reset(bool lightReset)
 {
   counters.notify = 0;
 
@@ -65,6 +65,12 @@ void ReadoutStats::reset()
   counters.timeframeIdFairMQ = 0;
 
   counters.firstOrbit = undefinedOrbit;
+
+  if (!lightReset) {
+    for (unsigned int i = 0; i < ReadoutStatsMaxItems; i++) {
+      counters.bufferUsage[i] = -1.0;
+    }
+  }
 }
 
 void ReadoutStats::print()