Merge pull request #217 from sy-c/master

v2.9.10

Author: sy-c

CMakeLists.txt

@@ -534,3 +534,9 @@ install(
 	DESTINATION ${CMAKE_INSTALL_PREFIX}/bin
 	RENAME o2-readout-status
 )
+
+install(
+	PROGRAMS src/readoutCommandLauncher.sh
+	DESTINATION ${CMAKE_INSTALL_PREFIX}/bin
+	RENAME o2-readout-command-launcher
+)

doc/configurationParameters.md

@@ -166,6 +166,9 @@ The parameters related to 3rd-party libraries are described here for convenience
 | readout | timeStop | string | | In standalone mode, time at which to execute stop. If not set, on int/term/quit signal. |
 | readout-monitor | broadcastHost | string | | used by readout-status to connect to readout-monitor broadcast channel. |
 | readout-monitor | broadcastPort | int | 0 | when set, the process will create a listening TCP port and broadcast statistics to connected clients. |
+| readout-monitor | logFile | string | | when set, the process will log received metrics to a file. |
+| readout-monitor | logFileHistory | int | 1 | defines the maximum number of previous log files to keep, when a maximum size is set. |
+| readout-monitor | logFileMaxSize | int | 128 | defines the maximum size of log file (in MB). When reaching this threshold, the log file is rotated. |
 | readout-monitor | monitorAddress | string | tcp://127.0.0.1:6008 | Address of the receiving ZeroMQ channel to receive readout statistics. |
 | readout-monitor | outputFormat | int | 0 | 0: default, human readable. 1: raw bytes. |
 | receiverFMQ | channelAddress | string | ipc:///tmp/pipe-readout | c.f. parameter with same name in consumer-FairMQchannel-* |

doc/howto.md

@@ -16,3 +16,22 @@ bytesMax=1G
 filesMax=0
 fileName=/tmp/data_%f.raw
 ```
+
+
+## How to launch custom commands on start/stop
+
+First, custom command shells must be enabled in the default readout settings `/etc/o2.d/readout-defaults.cfg`:
+```
+[readout]
+...
+customCommandsEnabled=1
+```
+This launches a sub-process shell on readout startup to execute commands later.
+
+Second, custom commands should be defined in the readout configuration file for associated state transitions: pre|post + START|STOP.
+For example, to start/stop internal ROC CTP emulator automatically with readout:
+```
+[readout]
+...
+customCommands=postSTART=roc-ctp-emulator --id=#1 --trigger-mode periodic --trigger-freq 40000 --hbmax 255 --bcmax 3563,preSTOP=o2-roc-ctp-emulator --id=#1 --eox
+```

doc/releaseNotes.md

@@ -439,3 +439,11 @@ This file describes the main feature changes for each readout.exe released versi
 
 ## v2.9.2 - 31/03/2022
 - Cosmetics improvements related to threads cleanup order on exit.
+
+## v2.10.0 - 13/04/2022
+- readout-monitor: added log file for received measurements.
+- added support for custom commands to be executed on states transitions.
+- Updated configuration parameters:
+  - readout.flushConsumerTimeout: when set, readout waits up to this amount of time that all data pages locked by consumers are released before stopping.
+- Added warning message on buffers low.
+- Added message at end of run showing the links which have provided data for each equipment and how much per link.

src/ConsumerFMQchannel.cxx

@@ -71,7 +71,7 @@ void incDataBlockStats(DataBlock* b)
     s->t0 = timeNowMicrosec();
     gReadoutStats.counters.pagesPendingFairMQ++;
     gReadoutStats.counters.notify++;
-    // printf("init %p\n",b);
+    // printf("init %p -> pages locked = %lu\n",b,(unsigned long)gReadoutStats.counters.pagesPendingFairMQ);
   }
 }
 
@@ -112,6 +112,12 @@ class ConsumerFMQchannel : public Consumer
 
   CounterStats repackSizeStats; // keep track of page size used when repacking
 
+  // custom log function for memory pool
+  void mplog(const std::string &msg) {
+    static InfoLogger::AutoMuteToken logMPToken(LogWarningSupport_(3230), 10, 60);
+    theLog.log(logMPToken, "Consumer %s : %s", name.c_str(), msg.c_str());
+  }
+
  public:
   std::vector<FairMQMessagePtr> messagesToSend; // collect HBF messages of each update
   uint64_t messagesToSendSize;                  // size (bytes) of messagesToSend payload
@@ -278,6 +284,8 @@ class ConsumerFMQchannel : public Consumer
     mp = theMemoryBankManager.getPagedPool(memoryPoolPageSize, memoryPoolNumberOfPages, memoryBankName);
     if (mp == nullptr) {
       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));
     }
     theLog.log(LogInfoDevel_(3008), "Using memory pool %d pages x %d bytes", memoryPoolNumberOfPages, memoryPoolPageSize);
   }

src/ConsumerStats.cxx

@@ -169,7 +169,7 @@ class ConsumerStats : public Consumer
       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());
 	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 );
+          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 );
 	}
       }
     }

src/MemoryPagesPool.cxx

@@ -24,6 +24,7 @@ MemoryPagesPool::MemoryPagesPool(size_t vPageSize, size_t vNumberOfPages, void*
   baseBlockAddress = vBaseAddress;
   baseBlockSize = vBaseSize;
   releaseBaseBlockCallback = vCallback;
+  state = BufferState::empty;
 
   // check page / header sizes
   assert(headerReservedSpace >= sizeof(DataBlock));
@@ -164,6 +165,9 @@ void* MemoryPagesPool::getPage()
     }
   }
 
+  // udpate buffer state
+  updateBufferState();
+
   return ptr;
 }
 
@@ -190,6 +194,9 @@ void MemoryPagesPool::releasePage(void* address)
 
   // put back page in list of available pages
   pagesAvailable->push(address);
+
+  // udpate buffer state
+  updateBufferState();
 }
 
 size_t MemoryPagesPool::getPageSize() { return pageSize; }
@@ -269,3 +276,30 @@ std::string MemoryPagesPool::getStats() {
   return "number of pages used: " + std::to_string(poolStats.getTotal()) + " average free pages: " + std::to_string((uint64_t)poolStats.getAverage()) + " minimum free pages: " + std::to_string(poolStats.getMinimum());
 }
 
+
+void MemoryPagesPool::setWarningCallback(const LogCallback& cb, double vthHigh, double vthOk) {
+  thHigh = vthHigh;
+  thOk = vthOk;
+  theLogCallback = cb;
+}
+
+void MemoryPagesPool::log(const std::string &msg) {
+  if (theLogCallback!=nullptr) {
+    // the log callback will format it with appropriate text headers (eg who owns the pool)
+    theLogCallback(msg);
+  }
+}
+
+void MemoryPagesPool::updateBufferState() {
+  double r = 1.0 - (getNumberOfPagesAvailable() * 1.0 / getTotalNumberOfPages());
+  if ((r == 1.0) && (state != BufferState::full)) {
+    state = BufferState::full;
+    log("buffer full");
+  } else if ((r > thHigh) && (state == BufferState::empty)) {
+    state = BufferState::high;
+    log("buffer usage is high");
+  } else if ((r < thOk) && ((state == BufferState::full) || (state == BufferState::high))) {
+    state = BufferState::empty;
+    log("buffer usage back to reasonable level");
+  }
+}

src/MemoryPagesPool.h

@@ -73,7 +73,21 @@ class MemoryPagesPool
 
   std::string getStats(); // return a string summarizing memory pool usage statistics
 
+  // an optional user-provided logging function for all memory pool related ops (including warnings on low)
+  typedef std::function<void(const std::string &)> LogCallback;
+
+  void setWarningCallback(const LogCallback& cb, double thHigh = 0.9, double thOk = 0.8);
+
  private:
+
+  LogCallback theLogCallback;
+  void log(const std::string &log);
+  double thHigh;
+  double thOk;
+  enum BufferState {empty, high, full};
+  BufferState state = BufferState::empty;
+  void updateBufferState();
+
   std::unique_ptr<AliceO2::Common::Fifo<void*>> pagesAvailable; // a buffer to keep track of individual pages
 
   size_t numberOfPages;                           // number of pages

src/ReadoutDatabase.cxx

@@ -106,9 +106,10 @@ ReadoutDatabase::ReadoutDatabase(const char* cx, int v, const LogCallback& cb) {
 }
 
 ReadoutDatabase::~ReadoutDatabase() {
-  if (db!=nullptr) {
+  if (db != nullptr) {
     mysql_close(db);
     db=nullptr;
+    log("DB closed");
   }
 }
 

src/ReadoutEquipment.cxx

@@ -48,7 +48,7 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
   cfg.getOptionalValue<double>("readout.rate", readoutRate, -1.0);
 
   // configuration parameter: | equipment-* | idleSleepTime | int | 200 | Thread idle sleep time, in microseconds. |
-  int cfgIdleSleepTime = 200;
+  cfgIdleSleepTime = 200;
   cfg.getOptionalValue<int>(cfgEntryPoint + ".idleSleepTime", cfgIdleSleepTime);
 
   // size of equipment output FIFO
@@ -201,6 +201,8 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
   if (mp == nullptr) {
     theLog.log(LogErrorSupport_(3230), "Failed to create pool of memory pages");
     throw __LINE__;
+  } else {
+    mp -> setWarningCallback(std::bind(&ReadoutEquipment::mplog, this, std::placeholders::_1));
   }
   // todo: move page align to MemoryPool class
   assert(pageSpaceReserved == mp->getPageSize() - mp->getDataBlockMaxSize());
@@ -227,6 +229,9 @@ void ReadoutEquipment::start()
     equipmentStats[i].reset();
     equipmentStatsLast[i] = 0;
   }
+  equipmentLinksUsed.reset();
+  equipmentLinksData.resize(RdhMaxLinkId + 1);
+  equipmentLinksData.clear();
   isError = 0;
   currentBlockId = 0;
   isDataOn = false;
@@ -280,6 +285,15 @@ void ReadoutEquipment::stop()
   theLog.log(LogInfoDevel_(3003), "Average pages pushed per iteration: %.1f", equipmentStats[EquipmentStatsIndexes::nBlocksOut].get() * 1.0 / (equipmentStats[EquipmentStatsIndexes::nLoop].get() - equipmentStats[EquipmentStatsIndexes::nIdle].get()));
   theLog.log(LogInfoDevel_(3003), "Average fifoready occupancy: %.1f", equipmentStats[EquipmentStatsIndexes::fifoOccupancyFreeBlocks].get() * 1.0 / (equipmentStats[EquipmentStatsIndexes::nLoop].get() - equipmentStats[EquipmentStatsIndexes::nIdle].get()));
   theLog.log(LogInfoDevel_(3003), "Average data throughput: %s", ReadoutUtils::NumberOfBytesToString(equipmentStats[EquipmentStatsIndexes::nBytesOut].get() / runningTime, "B/s").c_str());
+  theLog.log(LogInfoDevel_(3003), "Links used: %s", equipmentLinksUsed.to_string().c_str());
+
+  std::string perLinkStats;
+  for (unsigned int i = 0; i<= RdhMaxLinkId; i++) {
+    if (equipmentLinksUsed[i]) {
+      perLinkStats += "[" + std::to_string(i) + "]=" + NumberOfBytesToString(equipmentLinksData[i], "B", 1024) + " ";
+    }
+  }
+  theLog.log(LogInfoDevel_(3003), "Links data received: %s", perLinkStats.c_str());
 }
 
 ReadoutEquipment::~ReadoutEquipment()
@@ -678,6 +692,13 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
       cfgRdhDumpFirstInPageEnabled++;
     }
 
+    // update links statistics
+    if (h.getLinkId() <= RdhMaxLinkId) {
+      equipmentLinksUsed[h.getLinkId()] = 1;
+      equipmentLinksData[h.getLinkId()] += blockHeader.dataSize;
+    }
+
+
     // detect changes in detector bits field
     if (cfgRdhCheckDetectorField) {
       if (isDefinedLastDetectorField) {
@@ -830,3 +851,8 @@ void ReadoutEquipment::abortThread() {
   // ensure thread is stopped
   readoutThread = nullptr;
 }
+
+void ReadoutEquipment::mplog(const std::string &msg) {
+  static InfoLogger::AutoMuteToken logMPToken(LogWarningSupport_(3230), 10, 60);
+  theLog.log(logMPToken, "Equipment %s : %s", name.c_str(), msg.c_str());
+}