Merge pull request #226 from sy-c/master

v2.12.0

Author: sy-c

doc/configurationParameters.md

@@ -52,6 +52,7 @@ The parameters related to 3rd-party libraries are described here for convenience
 | consumer-data-sampling-* | address | string | ipc:///tmp/readout-pipe-1 | Address of the data sampling. |
 | 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).|
 | consumer-FairMQChannel-* | disableSending | int | 0 | If set, no data is output to FMQ channel. Used for performance test to create FMQ shared memory segment without pushing the data. |
+| consumer-FairMQChannel-* | enablePackedCopy | int | 0 | If set, the same superpage may be reused (space allowing) for the copy of multiple HBF (instead of a separate one for each copy). This allows a reduced memoryPoolNumberOfPages. |
 | 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.|
 | consumer-FairMQChannel-* | fmq-address | string | ipc:///tmp/pipe-readout | Address of the FMQ channel. Depends on transportType. c.f. FairMQ::FairMQChannel.h |
 | consumer-FairMQChannel-* | fmq-name | string | readout | Name of the FMQ channel. c.f. FairMQ::FairMQChannel.h |

doc/releaseNotes.md

@@ -471,3 +471,9 @@ This file describes the main feature changes for each readout.exe released versi
 
 ## v2.11.1 - 16/06/2022
 - Minor release for osx compatibility.
+
+## v2.12.0 - 09/08/2022
+- Updated configuration parameters:
+  - added consumer-FairMQChannel-*.enablePackedCopy. If set, the same superpage may be reused (space allowing) for the copy of multiple HBF (instead of a separate one for each copy). This allows a reduced memoryPoolNumberOfPages.
+- ROC fifo health check enabled, a warning message is logged when in/out ROC FIFOs are empty/full.
+

src/ConsumerFMQchannel.cxx

@@ -103,6 +103,7 @@ class ConsumerFMQchannel : public Consumer
   bool enableRawFormat = false;
   bool enableStfSuperpage = false; // optimized stf transport: minimize STF packets
   bool enableRawFormatDatablock = false;
+  int enablePackedCopy = 0; // default mode for repacking of page overlapping HBF. 0 = one page per copy, 1 = change page on TF only
 
   std::shared_ptr<MemoryBank> memBank; // a dedicated memory bank allocated by FMQ mechanism
   std::shared_ptr<MemoryPagesPool> mp; // a memory pool from which to allocate data pages
@@ -111,6 +112,11 @@ class ConsumerFMQchannel : public Consumer
   int memoryPoolNumberOfPages;
 
   CounterStats repackSizeStats; // keep track of page size used when repacking
+  uint64_t nPagesUsedForRepack = 0; // count pages used for repack
+  uint64_t nPagesUsedInput = 0; // count pages received
+
+  DataBlockContainerReference copyBlockBuffer = nullptr; // current buffer used for packed copy
+  uint64_t lastTimeframeId = undefinedTimeframeId; // keep track of latest TF id received
 
   // custom log function for memory pool
   void mplog(const std::string &msg) {
@@ -245,6 +251,7 @@ class ConsumerFMQchannel : public Consumer
         if (hint != nullptr) {
           DataBlockContainerReference* blockRef = (DataBlockContainerReference*)hint;
           //printf("ack hint=%p page %p\n",hint,(*blockRef)->getData());
+	  //printf("ptr %p: use_count=%d\n",blockRef, (int)blockRef->use_count());
           decDataBlockStats((*blockRef)->getData());
           delete blockRef;
         }
@@ -291,17 +298,22 @@ class ConsumerFMQchannel : public Consumer
       }
     }
     theLog.log(LogInfoDevel_(3008), "Using memory pool [%d]: %d pages x %d bytes", mp->getId(), memoryPoolNumberOfPages, memoryPoolPageSize);
+
+    // configuration parameter: | consumer-FairMQChannel-* | enablePackedCopy | int | 0 | If set, the same superpage may be reused (space allowing) for the copy of multiple HBF (instead of a separate one for each copy). This allows a reduced memoryPoolNumberOfPages. |
+    cfg.getOptionalValue<int>(cfgEntryPoint + ".enablePackedCopy", enablePackedCopy);
+    theLog.log(LogInfoDevel_(3008), "Packed copy enabled = %d", enablePackedCopy);
   }
 
   ~ConsumerFMQchannel()
   {
     // log memory pool statistics
     if (mp!=nullptr) {
       theLog.log(LogInfoDevel_(3003), "Consumer %s - memory pool statistics ... %s", name.c_str(), mp->getStats().c_str());
-      theLog.log(LogInfoDevel_(3003), "Consumer %s - STFB repacking statistics ... number: %" PRIu64 " average page size: %" PRIu64 " max page size: %" PRIu64, name.c_str(), repackSizeStats.getCount(), (uint64_t)repackSizeStats.getAverage(), repackSizeStats.getMaximum());
+      theLog.log(LogInfoDevel_(3003), "Consumer %s - STFB repacking statistics ... number: %" PRIu64 " average page size: %" PRIu64 " max page size: %" PRIu64 " repacked/received = %" PRIu64 "/%" PRIu64 " = %.1f%%", name.c_str(), repackSizeStats.getCount(), (uint64_t)repackSizeStats.getAverage(), repackSizeStats.getMaximum(), nPagesUsedForRepack, nPagesUsedInput, nPagesUsedForRepack * 100.0 / nPagesUsedInput);
     }
 
     // release in reverse order
+    copyBlockBuffer = nullptr;
     mp = nullptr;
     memoryBuffer = nullptr; // warning: data range may still be referenced in memory bank manager
     sendingChannel = nullptr;
@@ -317,6 +329,8 @@ class ConsumerFMQchannel : public Consumer
   int pushData(DataSetReference& bc)
   {
 
+    nPagesUsedInput += bc->size();
+
     if (disableSending) {
       totalPushSuccess++;
       return 0;
@@ -498,6 +512,15 @@ class ConsumerFMQchannel : public Consumer
       // set flag when this is last STF in timeframe
       if (b->header.flagEndOfTimeframe) {
         stfHeader->lastTFMessage = 1;
+	//printf("end of TF %d eq %d link %d\n ", (int) b->header.timeframeId, (int) b->header.equipmentId, (int)b->header.linkId);
+	copyBlockBuffer = nullptr;
+      }
+
+      // detect changes of TF id
+      if (b->header.timeframeId != lastTimeframeId) {
+        lastTimeframeId = b->header.timeframeId;
+	copyBlockBuffer = nullptr;
+        //printf("TF %d start\n", (int) lastTimeframeId);
       }
 
       if (isFirst) {
@@ -635,7 +658,26 @@ class ConsumerFMQchannel : public Consumer
         }
         DataBlockContainerReference copyBlock = nullptr;
         try {
-          copyBlock = mp->getNewDataBlockContainer();
+	  if (enablePackedCopy) {
+	    for (int i = 0; i<=2; i++) {
+              // allocate new buffer for copies if needed
+	      if (copyBlockBuffer == nullptr) {
+	        copyBlockBuffer = mp->getNewDataBlockContainer();
+		nPagesUsedForRepack++;
+		continue;
+	      }
+	      // try to allocate sub-block
+	      copyBlock = DataBlockContainer::getChildBlock(copyBlockBuffer, totalSize);
+	      if (copyBlock == nullptr) {
+	        copyBlockBuffer = nullptr;
+		continue;
+	      }
+	      break;
+	    }
+	  } else {
+            copyBlock = mp->getNewDataBlockContainer();
+	    nPagesUsedForRepack++;
+	  }
         } catch (...) {
         }
         if (copyBlock == nullptr) {

src/DataBlockContainer.h

@@ -18,11 +18,19 @@
 #include <stdint.h>
 #include <stdlib.h>
 
+//#define DATABLOCKCONTAINER_DEBUG
+#ifdef DATABLOCKCONTAINER_DEBUG
+#include <stdio.h>
+#endif
+
 #include "DataBlock.h"
 
 // A container class for data blocks.
 // In particular, allows to take care of the block release after use.
 
+class DataBlockContainer;
+using DataBlockContainerReference = std::shared_ptr<DataBlockContainer>;
+
 class DataBlockContainer
 {
 
@@ -42,6 +50,9 @@ class DataBlockContainer
     if (releaseCallback != nullptr) {
       releaseCallback();
     }
+    #ifdef DATABLOCKCONTAINER_DEBUG
+      printf("Releasing DataBlockContainer @ %p\n", data);
+    #endif
   };
 
   DataBlock* getData()
@@ -54,10 +65,44 @@ class DataBlockContainer
     return dataBufferSize;
   };
 
+
+  static DataBlockContainerReference getChildBlock(DataBlockContainerReference parentBlock, uint64_t v_dataBufferSizeNeeded, uint64_t roundUp = 0) {
+    uint64_t bufferSize = v_dataBufferSizeNeeded + sizeof(DataBlock);
+    if (roundUp) {
+      uint64_t r = bufferSize % roundUp;
+      if (r) {
+        bufferSize += roundUp - r;
+      }
+    }
+    if (bufferSize > parentBlock->getData()->header.dataSize - parentBlock->dataBufferUsed) {
+      return nullptr;
+    }
+    DataBlock *b = (DataBlock *)&((char *)parentBlock->getData()->data)[parentBlock->dataBufferUsed];
+    b->header = defaultDataBlockHeader;
+    b->header.dataSize = v_dataBufferSizeNeeded;
+    b->data = &(((char*)b)[sizeof(DataBlock)]);
+
+    #ifdef DATABLOCKCONTAINER_DEBUG
+    printf("Block %p - creating child @ %p - payload size %d\n", parentBlock->getData(), b, (int)v_dataBufferSizeNeeded);
+    #endif
+
+    DataBlockContainerReference childBlock = std::make_shared<DataBlockContainer>(b, bufferSize);
+    if (childBlock == nullptr) {
+      return nullptr;
+    }
+    parentBlock->dataBufferUsed += bufferSize;
+    childBlock->parentBlock = parentBlock;
+    return childBlock;
+  };
+
+
  protected:
   DataBlock* data;                 // The DataBlock in use
   uint64_t dataBufferSize = 0;     // Usable memory size pointed by data. Unspecified if zero.
   ReleaseCallback releaseCallback; // Function called on object destroy, to release dataBlock.
+
+  DataBlockContainerReference parentBlock = nullptr; // reference to parent block used, if any
+  uint64_t dataBufferUsed = 0; // size of buffer used for child blocks.
 };
 
 #endif

src/DataSet.h

@@ -17,7 +17,6 @@
 // A container to store a vector of shared pointers to data block containers.
 // Can typically be used to pass a set of blocks to a function.
 
-using DataBlockContainerReference = std::shared_ptr<DataBlockContainer>;
 using DataSet = std::vector<DataBlockContainerReference>;
 using DataSetReference = std::shared_ptr<DataSet>;
 

src/MemoryPagesPool.cxx

@@ -298,6 +298,9 @@ void MemoryPagesPool::log(const std::string &msg) {
 }
 
 void MemoryPagesPool::updateBufferState() {
+  if (theLogCallback == nullptr) {
+    return;
+  }
   double r = 1.0 - (getNumberOfPagesAvailable() * 1.0 / getTotalNumberOfPages());
   if ((r == 1.0) && (state != BufferState::full)) {
     state = BufferState::full;

src/ReadoutEquipmentRORC.cxx

@@ -15,6 +15,7 @@
 #include <ReadoutCard/Exception.h>
 #include <ReadoutCard/MemoryMappedFile.h>
 #include <ReadoutCard/Parameters.h>
+#include <Common/SimpleLog.h>
 #include <cstring>
 #include <mutex>
 #include <string>
@@ -64,6 +65,11 @@ class ReadoutEquipmentRORC : public ReadoutEquipment
 
   int32_t lastPacketDropped = 0;             // latest value of CRU packet dropped counter
   AliceO2::Common::Timer packetDroppedTimer; // a timer to set period of packet dropped checks
+
+  SimpleLog logRocCalls; // ROC debug log file, for ROC library traces
+  bool logRocCallsEnable = false; // flag to enable ROC debug log file
+  std::string logRocCallsPath = "/tmp/roclog.txt"; // path to ROC debug log file
+  unsigned long logRocCallsMaxFileSize = 10000000; // max size (bytes) of ROC debug log file
 };
 
 // std::mutex readoutEquipmentRORCLock;
@@ -174,6 +180,11 @@ ReadoutEquipmentRORC::ReadoutEquipmentRORC(ConfigFile& cfg, std::string name) :
 
     // todo: log parameters ?
 
+    if (logRocCallsEnable) {
+      logRocCalls.setLogFile(logRocCallsPath.c_str(), logRocCallsMaxFileSize);
+      logRocCalls.info("Logs of ROC library calls by readout process");
+    }
+
   } catch (const std::exception& e) {
     theLog.log(LogErrorSupport_(3240), "Exception : %s", e.what());
     theLog.log(LogErrorSupport_(3240), "%s", boost::diagnostic_information(e).c_str());
@@ -216,12 +227,10 @@ Thread::CallbackResult ReadoutEquipmentRORC::prepareBlocks()
     } 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());
       }
-      }
     }
   }
 
@@ -267,6 +276,9 @@ Thread::CallbackResult ReadoutEquipmentRORC::prepareBlocks()
     }
   }
   equipmentStats[EquipmentStatsIndexes::nPushedUp].increment(nPushed);
+  if (logRocCallsEnable) {
+    logRocCalls.info("pushed %d",nPushed);
+  }
 
   // check fifo occupancy ready queue size for stats
   equipmentStats[EquipmentStatsIndexes::fifoOccupancyReadyBlocks].set(channel->getReadyQueueSize());
@@ -284,6 +296,10 @@ Thread::CallbackResult ReadoutEquipmentRORC::prepareBlocks()
 
   // this is to be called periodically for driver internal business
   channel->fillSuperpages();
+  if (logRocCallsEnable) {
+    logRocCalls.info("fillSuperpage()");
+  }
+
 
   // readoutEquipmentRORCLock.unlock();
 
@@ -317,6 +333,9 @@ DataBlockContainerReference ReadoutEquipmentRORC::getNextBlock()
   for(int loop = 0; loop < maxLoop; loop++) {
     bool doLoop = 0;
     if ((channel->getReadyQueueSize() > 0)) {
+      if (logRocCallsEnable) {
+        logRocCalls.info("pop (%d available)", channel->getReadyQueueSize());
+      }
       // get next page from FIFO
       auto superpage = channel->popSuperpage();
       void* mpPageAddress = (void*)(superpage.getUserData());
@@ -370,6 +389,9 @@ void ReadoutEquipmentRORC::setDataOn()
     // start DMA
     theLog.log(LogInfoDevel_(3010), "Starting DMA for ROC %s", getName().c_str());
     channel->startDma();
+    if (logRocCallsEnable) {
+      logRocCalls.info("startdma()");
+    }
 
     // get FIFO depth (it should be fully empty when starting)
     // can not be done before startDma() - would return 0
@@ -415,6 +437,9 @@ void ReadoutEquipmentRORC::setDataOff()
     theLog.log(LogInfoDevel_(3010), "Stopping DMA for ROC %s", getName().c_str());
     try {
       channel->stopDma();
+      if (logRocCallsEnable) {
+        logRocCalls.info("stopdma()");
+      }
     } catch (const std::exception& e) {
       theLog.log(LogErrorSupport_(3240), "Exception : %s", e.what());
       theLog.log(LogErrorSupport_(3240), "%s", boost::diagnostic_information(e).c_str());

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.11.1"
+#define READOUT_VERSION "2.12.0"
 

src/mainReadout.cxx

@@ -565,6 +565,26 @@ int Readout::configure(const boost::property_tree::ptree& properties)
   };
   mergeConfig(cfg.get(), properties);
 
+  // extract optional configuration parameters
+  // configuration parameter: | readout | customCommands | string | | List of key=value pairs defining some custom shell commands to be executed at before/after state change commands. |
+  if (customCommandsShellPid) {
+    std::string cfgCustomCommandsList;
+    customCommands.clear();
+    cfg.getOptionalValue<std::string>("readout.customCommands", cfgCustomCommandsList);
+    if (getKeyValuePairsFromString(cfgCustomCommandsList, customCommands)) {
+      theLog.log(LogWarningDevel_(3102), "Failed to parse custom commands");
+      customCommands.clear();
+    } else {
+      theLog.log(LogInfoDevel_(3013), "Registered custom commands:");
+      for (const auto &kv : customCommands) {
+	theLog.log(LogInfoDevel_(3013), "%s : %s", kv.first.c_str(), kv.second.c_str());
+      }
+    }
+  }
+
+  // execute custom command
+  executeCustomCommand("preCONFIGURE");
+
   // try to prevent deep sleeps
   bool deepsleepDisabled = false;
   int maxLatency = 2;
@@ -584,23 +604,6 @@ int Readout::configure(const boost::property_tree::ptree& properties)
     theLog.log(LogInfoDevel, "CPU deep sleep not disabled for process");
   }
 
-  // extract optional configuration parameters
-  // configuration parameter: | readout | customCommands | string | | List of key=value pairs defining some custom shell commands to be executed at before/after state change commands. |
-  if (customCommandsShellPid) {
-    std::string cfgCustomCommandsList;
-    customCommands.clear();
-    cfg.getOptionalValue<std::string>("readout.customCommands", cfgCustomCommandsList);
-    if (getKeyValuePairsFromString(cfgCustomCommandsList, customCommands)) {
-      theLog.log(LogWarningDevel_(3102), "Failed to parse custom commands");
-      customCommands.clear();
-    } else {
-      theLog.log(LogInfoDevel_(3013), "Registered custom commands:");
-      for (const auto &kv : customCommands) {
-	theLog.log(LogInfoDevel_(3013), "%s : %s", kv.first.c_str(), kv.second.c_str());
-      }
-    }
-  }
-
   // configuration parameter: | readout | exitTimeout | double | -1 | Time in seconds after which the program exits automatically. -1 for unlimited. |
   cfgExitTimeout = -1;
   cfg.getOptionalValue<double>("readout.exitTimeout", cfgExitTimeout);
@@ -1027,6 +1030,9 @@ int Readout::configure(const boost::property_tree::ptree& properties)
   }
   theLog.log(LogInfoDevel, "Aggregator: %d equipments", nEquipmentsAggregated);
 
+  // execute custom command
+  executeCustomCommand("postCONFIGURE");
+
   theLog.log(LogInfoSupport_(3005), "Readout completed CONFIGURE");
   gReadoutStats.counters.state = stringToUint64("ready");
   gReadoutStats.counters.notify++;

src/readoutConfigEditor.tcl

@@ -30,6 +30,7 @@ set configurationParametersDescriptor {
 | consumer-data-sampling-* | address | string | ipc:///tmp/readout-pipe-1 | Address of the data sampling. |
 | 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).|
 | consumer-FairMQChannel-* | disableSending | int | 0 | If set, no data is output to FMQ channel. Used for performance test to create FMQ shared memory segment without pushing the data. |
+| consumer-FairMQChannel-* | enablePackedCopy | int | 0 | If set, the same superpage may be reused (space allowing) for the copy of multiple HBF (instead of a separate one for each copy). This allows a reduced memoryPoolNumberOfPages. |
 | 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.|
 | consumer-FairMQChannel-* | fmq-address | string | ipc:///tmp/pipe-readout | Address of the FMQ channel. Depends on transportType. c.f. FairMQ::FairMQChannel.h |
 | consumer-FairMQChannel-* | fmq-name | string | readout | Name of the FMQ channel. c.f. FairMQ::FairMQChannel.h |