Merge pull request #238 from sy-c/master

update

Author: sy-c

CMakeLists.txt

@@ -239,6 +239,7 @@ target_include_directories(objReadoutAggregator PRIVATE ${READOUT_INCLUDE_DIRS})
 # utilities, including memory handling
 add_library(
         objReadoutUtils OBJECT
+        ${SOURCE_DIR}/readoutInfoLogger.cxx
         ${SOURCE_DIR}/ReadoutUtils.cxx
         ${SOURCE_DIR}/RdhUtils.cxx
         ${SOURCE_DIR}/CounterStats.cxx

doc/configurationParameters.md

@@ -69,7 +69,8 @@ The parameters related to 3rd-party libraries are described here for convenience
 | consumer-FairMQChannel-* | unmanagedMemorySize | bytes |  | Size of the memory region to be created. c.f. FairMQ::FairMQUnmanagedRegion.h. If not set, no special FMQ memory region is created. |
 | consumer-fileRecorder-* | bytesMax | bytes | 0 | Maximum number of bytes to write to each file. Data pages are never truncated, so if writing the full page would exceed this limit, no data from that page is written at all and file is closed. If zero (default), no maximum size set.|
 | consumer-fileRecorder-* | dataBlockHeaderEnabled | int | 0 | Enable (1) or disable (0) the writing to file of the internal readout header (Readout DataBlock.h) between the data pages, to easily navigate through the file without RDH decoding. If disabled, the raw data pages received from CRU are written without further formatting. |
-| consumer-fileRecorder-* | dropEmptyHBFrames | int | 0 | If 1, memory pages are scanned and empty HBframes are discarded, i.e. couples of packets which contain only RDH, the first one with pagesCounter=0 and the second with stop bit set. This setting does not change the content of in-memory data pages, other consumers would still get full data pages with empty packets. This setting is meant to reduce the amount of data recorded for continuous detectors in triggered mode.|
+| consumer-fileRecorder-* | dropEmptyHBFrames | int | 0 | If 1, memory pages are scanned and empty HBframes are discarded, i.e. couples of packets which contain only RDH, the first one with pagesCounter=0 and the second with stop bit set. This setting does not change the content of in-memory data pages, other consumers would still get full data pages with empty packets. This setting is meant to reduce the amount of data recorded for continuous detectors in triggered mode. Use with dropEmptyHBFramesTriggerMask, if some empty frames with specific trigger types need to be kept (eg TF or SOC). |
+| consumer-fileRecorder-* | dropEmptyHBFramesTriggerMask | int | 0 | (when using dropEmptyHBFrames = 1) empty HB frames are kept if any bit in RDH TriggerType field matches this pattern (RDHTriggerType & TriggerMask != 0). To be provided as a decimal value: eg 2048 (TF triggers, bit 11), 3584 (TF + SOC + EOC bits 9,10,11). |
 | consumer-fileRecorder-* | fileName | string | | Path to the file where to record data. The following variables are replaced at runtime: ${XXX} -> get variable XXX from environment, %t -> unix timestamp (seconds since epoch), %T -> formatted date/time, %i -> equipment ID of each data chunk (used to write data from different equipments to different output files), %l -> link ID (used to write data from different links to different output files). |
 | consumer-fileRecorder-* | filesMax | int | 1 | If 1 (default), file splitting is disabled: file is closed whenever a limit is reached on a given recording stream. Otherwise, file splitting is enabled: whenever the current file reaches a limit, it is closed an new one is created (with an incremental name). If <=0, an unlimited number of incremental chunks can be created. If non-zero, it defines the maximum number of chunks. The file name is suffixed with chunk number (by default, ".001, .002, ..." at the end of the file name. One may use "%f" in the file name to define where this incremental file counter is printed. |
 | consumer-fileRecorder-* | pagesMax | int | 0 | Maximum number of data pages accepted by recorder. If zero (default), no maximum set.|

doc/releaseNotes.md

@@ -504,3 +504,10 @@ This file describes the main feature changes for each readout.exe released versi
 
 ## v2.15.2 - 08/12/2022
 - Added ROC link Id for some log messages, as provided from new superpage metadata link field from ROC library.
+
+## next version
+- Superpage internal metadata not stored at beginning of page but separately in heap, by default. Behavior can be adjusted with MemoryPool::headerReservedSpace.
+- equipment-CruEmulator: set TF trigger type (bit 11) on new TF. Used for dropEmptyHBFramesTriggerMask tests.
+- Updated configuration parameters:
+  - added consumer-fileRecorder-*.dropEmptyHBFramesTriggerMask: when using dropEmptyHBFrames = 1, keep some empty HB frames with trigger type matching given (decimal) mask.
+- Added support for RDHv7. Readout already accepts data from equipments with v7, but still generates data (internal sw generator) with v6, for compatibility with current software downstream.

src/ConsumerFMQchannel.cxx

@@ -9,6 +9,9 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
+#define ENABLE_LOG_CODEWRONG
+#include "readoutInfoLogger.h"
+
 #include "Consumer.h"
 #include "MemoryBank.h"
 #include "MemoryBankManager.h"
@@ -62,8 +65,13 @@ static_assert(sizeof(DataBlockFMQStats) <= DataBlockHeaderUserSpace, "DataBlockF
 //uint64_t ddsizemem=0;
 
 
-void initDataBlockStats(DataBlock* b, uint64_t v_memorySizeAccounted = 0)
+void initDataBlockStats(DataBlockContainerReference* blockRef, uint64_t v_memorySizeAccounted = 0)
 {
+  if (blockRef == nullptr) {return;}
+  if (*blockRef == nullptr) {return;}
+  DataBlock* b = (*blockRef)->getData();
+  if (b == nullptr) {return;}
+  if ((*blockRef)->isChildBlock()) {LOG_CODEWRONG; return;}
   DataBlockFMQStats* s = (DataBlockFMQStats*)&(b->header.userSpace);
   s->magic = 0xAA;
   s->countRef = 0;
@@ -72,30 +80,55 @@ void initDataBlockStats(DataBlock* b, uint64_t v_memorySizeAccounted = 0)
   //printf ("TF %d adding mem sz %d\n", (int)b->header.timeframeId, (int) v_memorySizeAccounted);
 }
 
-void incDataBlockStats(DataBlock* b, uint64_t dataSizeAccounted = 0)
+void incDataBlockStats(DataBlockContainerReference* blockRef, uint64_t dataSizeAccounted = 0)
 {
-  //printf("inc %p\n",b);
+  if (blockRef == nullptr) {return;}
+  if (*blockRef == nullptr) {return;}
+  DataBlockContainerReference parentBlock = nullptr;
+  if ((*blockRef)->isChildBlock()) {
+    // for a child block, update stats of the parent block
+    parentBlock = (*blockRef)->getParent();
+    blockRef = &parentBlock;
+  }
+  DataBlock* b = (*blockRef)->getData();
+  if (b == nullptr) {return;}
+  //printf("inc %p\n",b->data);
   DataBlockFMQStats* s = (DataBlockFMQStats*)&(b->header.userSpace);
+
   if (s->magic != 0xAA)
     return;
   if ((s->countRef++) == 0) {
     s->t0 = timeNowMicrosec();
     gReadoutStats.counters.pagesPendingFairMQ++;
     gReadoutStats.counters.notify++;
-    // printf("init %p -> pages locked = %lu\n",b,(unsigned long)gReadoutStats.counters.pagesPendingFairMQ);
+    // printf("init %p -> pages locked = %lu\n",b->data,(unsigned long)gReadoutStats.counters.pagesPendingFairMQ);
     gReadoutStats.counters.ddMemoryPendingBytes += s->memorySizeAccounted;
     //printf("adding %d / %d\n", (int)dataSizeAccounted, (int)s->memorySizeAccounted);
     //ddsizemem+=s->memorySizeAccounted;
+    //printf("page %p pool %p\n",b->data,(*blockRef)->memoryPagesPoolPtr);
+    updatePageStateFromDataBlockContainerReference(*blockRef, MemoryPage::PageState::InFMQ);
   }
   s->dataSizeAccounted += dataSizeAccounted;
   gReadoutStats.counters.ddPayloadPendingBytes += dataSizeAccounted;
   //ddsizepayload += dataSizeAccounted;
 }
 
-void decDataBlockStats(DataBlock* b)
+void decDataBlockStats(DataBlockContainerReference* blockRef)
 {
+  if (blockRef == nullptr) {return;}
+  if (*blockRef == nullptr) {return;}
+  DataBlockContainerReference parentBlock = nullptr;
+  if ((*blockRef)->isChildBlock()) {
+    // for a child block, update stats of the parent block
+    parentBlock = (*blockRef)->getParent();
+    blockRef = &parentBlock;
+  }
+  DataBlock* b = (*blockRef)->getData();
+  if ((*blockRef)->isChildBlock()) {
+    b = (*blockRef)->getParent()->getData();
+  }
+  if (b == nullptr) {return;}
   DataBlockFMQStats* s = (DataBlockFMQStats*)&(b->header.userSpace);
-  //printf("dec %p\n",b);
   if (s->magic != 0xAA)
     return;
   if ((--s->countRef) == 0) {
@@ -317,7 +350,7 @@ class ConsumerFMQchannel : public Consumer
           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());
+          decDataBlockStats(blockRef);
           delete blockRef;
         }
       },fair::mq::RegionConfig{false,false});  // lock / zero - done later
@@ -445,6 +478,7 @@ class ConsumerFMQchannel : public Consumer
           totalPushError++;
           return -1;
         }
+        (*blockRef)->memoryPagesPoolPtr = br->memoryPagesPoolPtr; // keep ref to memoryPagesPool for state updates
         void* hint = (void*)blockRef;
         void* blobPtr = b->data;
         size_t blobSize = (size_t)b->header.dataSize;
@@ -478,6 +512,7 @@ class ConsumerFMQchannel : public Consumer
           totalPushError++;
           return -1;
         }
+        (*ptr)->memoryPagesPoolPtr = br->memoryPagesPoolPtr; // keep ref to memoryPagesPool for state updates
         std::unique_ptr<FairMQMessage> msgHeader(transportFactory->CreateMessage((void*)&(br->getData()->header), (size_t)(br->getData()->header.headerSize), msgcleanupCallback, (void*)nullptr));
         std::unique_ptr<FairMQMessage> msgBody(transportFactory->CreateMessage((void*)(br->getData()->data), (size_t)(br->getData()->header.dataSize), msgcleanupCallback, (void*)(ptr)));
 
@@ -505,6 +540,7 @@ class ConsumerFMQchannel : public Consumer
         totalPushError++;
         return -1;
       }
+      (*blockRef)->memoryPagesPoolPtr = headerBlock->memoryPagesPoolPtr; // keep ref to memoryPagesPool for state updates
       SubTimeframe* stfHeader = (SubTimeframe*)headerBlock->getData()->data;
       if (stfHeader == nullptr) {
         totalPushError++;
@@ -550,6 +586,7 @@ class ConsumerFMQchannel : public Consumer
           totalPushError++;
           return -1;
         }
+        (*blockRef)->memoryPagesPoolPtr = br->memoryPagesPoolPtr; // keep ref to memoryPagesPool for state updates
         void* hint = (void*)blockRef;
         void* blobPtr = b->data;
         size_t blobSize = (size_t)b->header.dataSize;
@@ -728,6 +765,7 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
     totalPushError++;
     return -1;
   }
+  (*blockRef)->memoryPagesPoolPtr = headerBlock->memoryPagesPoolPtr; // keep ref to memoryPagesPool for state updates
   SubTimeframe* stfHeader = (SubTimeframe*)headerBlock->getData()->data;
   if (stfHeader == nullptr) {
     totalPushError++;
@@ -820,8 +858,8 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
   assert(ddm.messagesToSend.empty());
   if (memoryBuffer) {
     // printf("send H %p\n", blockRef);
-    initDataBlockStats((*blockRef)->getData(), headerBlock->getDataBufferSize());
-    incDataBlockStats((*blockRef)->getData(), sizeof(SubTimeframe));
+    initDataBlockStats(blockRef, headerBlock->getDataBufferSize());
+    incDataBlockStats(blockRef, sizeof(SubTimeframe));
     ddm.messagesToSend.emplace_back(sendingChannel->NewMessage(memoryBuffer, (void*)stfHeader, sizeof(SubTimeframe), (void*)(blockRef)));
   } else {
     ddm.messagesToSend.emplace_back(sendingChannel->NewMessage((void*)stfHeader, sizeof(SubTimeframe), msgcleanupCallback, (void*)(blockRef)));
@@ -849,6 +887,10 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
     pf.HBid = id;
     // create a copy of the reference, in a newly allocated object, so that reference is kept alive until this new object is destroyed in the cleanupCallback
     pf.blockRef = new DataBlockContainerReference(br);
+    if (pf.blockRef == nullptr) {
+      throw __LINE__;
+    }
+    (*pf.blockRef)->memoryPagesPoolPtr = br->memoryPagesPoolPtr; // keep ref to memoryPagesPool for state updates
     pendingFrames.push_back(pf);
   };
 
@@ -874,7 +916,7 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
       // create and queue a fmq message
       if (memoryBuffer) {
         // printf("send D %p\n", hint);
-        incDataBlockStats((*(pendingFrames[0].blockRef))->getData(), l);
+        incDataBlockStats(pendingFrames[0].blockRef, l);
 //        printf("mem1 sz = %d\n",(int)(*(pendingFrames[0].blockRef))->getData()->header.memorySize);
         ddm.messagesToSend.emplace_back(sendingChannel->NewMessage(memoryBuffer, (void*)(&(b->data[ix])), (size_t)(l), hint));
       } else {
@@ -917,6 +959,7 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
 	      isNewBlock = 1;
               if (copyBlockBuffer != nullptr) {
 	        copyBlockMemSize = copyBlockBuffer->getDataBufferSize();
+                initDataBlockStats(&copyBlockBuffer, copyBlockMemSize);
 	      }
 	      nPagesUsedForRepack++;
 	      continue;
@@ -934,6 +977,7 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
 	  isNewBlock = 1;
 	  if (copyBlock != nullptr) {
 	    copyBlockMemSize = copyBlock->getDataBufferSize();
+            initDataBlockStats(&copyBlock, copyBlockMemSize);
 	  }
 	  nPagesUsedForRepack++;
 	}
@@ -951,6 +995,10 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
       }
       auto blockRef = new DataBlockContainerReference(copyBlock);
       char* newBlock = (char*)copyBlock->getData()->data;
+      if (blockRef ==nullptr) {
+        throw __LINE__;
+      }
+      (*blockRef)->memoryPagesPoolPtr = copyBlock->memoryPagesPoolPtr; // keep ref to memoryPagesPool for state updates
 
       int newIx = 0;
       for (auto& f : pendingFrames) {
@@ -972,8 +1020,7 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
       // create and queue a fmq message
       if (memoryBuffer) {
         // printf("send D2 %p\n", blockRef);
-        initDataBlockStats((*blockRef)->getData(), copyBlockMemSize);
-        incDataBlockStats((*blockRef)->getData(), totalSize);
+        incDataBlockStats(blockRef, totalSize);
         ddm.messagesToSend.emplace_back(sendingChannel->NewMessage(memoryBuffer, (void*)newBlock, totalSize, (void*)(blockRef)));
       } else {
         ddm.messagesToSend.emplace_back(sendingChannel->NewMessage((void*)newBlock, totalSize, msgcleanupCallback, (void*)(blockRef)));
@@ -988,7 +1035,7 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
   try {
     for (auto& br : *bc) {
       DataBlock* b = br->getData();
-      initDataBlockStats(b, br->getDataBufferSize());
+      initDataBlockStats(&br, br->getDataBufferSize());
 
       unsigned int HBstart = 0;
       for (int offset = 0; offset + sizeof(o2::Header::RAWDataHeader) <= b->header.dataSize;) {

src/ConsumerFileRecorder.cxx

@@ -211,7 +211,7 @@ class ConsumerFileRecorder : public Consumer
       }
     }
 
-    // configuration parameter: | consumer-fileRecorder-* | dropEmptyHBFrames | int | 0 | If 1, memory pages are scanned and empty HBframes are discarded, i.e. couples of packets which contain only RDH, the first one with pagesCounter=0 and the second with stop bit set. This setting does not change the content of in-memory data pages, other consumers would still get full data pages with empty packets. This setting is meant to reduce the amount of data recorded for continuous detectors in triggered mode.|
+    // configuration parameter: | consumer-fileRecorder-* | dropEmptyHBFrames | int | 0 | If 1, memory pages are scanned and empty HBframes are discarded, i.e. couples of packets which contain only RDH, the first one with pagesCounter=0 and the second with stop bit set. This setting does not change the content of in-memory data pages, other consumers would still get full data pages with empty packets. This setting is meant to reduce the amount of data recorded for continuous detectors in triggered mode. Use with dropEmptyHBFramesTriggerMask, if some empty frames with specific trigger types need to be kept (eg TF or SOC). |
     cfg.getOptionalValue(cfgEntryPoint + ".dropEmptyHBFrames", dropEmptyHBFrames, 0);
     if (dropEmptyHBFrames) {
       if (recordWithDataBlockHeader) {
@@ -220,6 +220,13 @@ class ConsumerFileRecorder : public Consumer
       }
       theLog.log(LogInfoSupport_(3002), "Some packets with RDH-only payload will not be recorded to file, option dropEmptyHBFrames is enabled");
     }
+
+    // configuration parameter: | consumer-fileRecorder-* | dropEmptyHBFramesTriggerMask | int | 0 | (when using dropEmptyHBFrames = 1) empty HB frames are kept if any bit in RDH TriggerType field matches this pattern (RDHTriggerType & TriggerMask != 0). To be provided as a decimal value: eg 2048 (TF triggers, bit 11), 3584 (TF + SOC + EOC bits 9,10,11). |
+    cfg.getOptionalValue(cfgEntryPoint + ".dropEmptyHBFramesTriggerMask", dropEmptyHBFramesTriggerMask, 0);
+    if ((dropEmptyHBFrames)&&(dropEmptyHBFramesTriggerMask)) {
+      theLog.log(LogInfoSupport_(3002), "Some packets with RDH-only payload will be recorded when their trigger type matches mask 0x%X", dropEmptyHBFramesTriggerMask);
+    }
+
   }
 
   ~ConsumerFileRecorder() {}
@@ -504,14 +511,14 @@ class ConsumerFileRecorder : public Consumer
     };
 
     auto isEmptyHBstop = [&](RdhHandle& h) {
-      if ((h.getStopBit()) && (h.getHeaderSize() == h.getMemorySize())) {
+      if ((h.getStopBit()) && (h.getHeaderSize() == h.getMemorySize()) && ((h.getTriggerType() & (uint32_t)dropEmptyHBFramesTriggerMask) == 0) ) {
         return true;
       }
       return false;
     };
 
     auto isEmptyHBstart = [&](RdhHandle& h) {
-      if ((h.getPagesCounter() == 0) && (h.getHeaderSize() == h.getMemorySize())) {
+      if ((h.getPagesCounter() == 0) && (h.getHeaderSize() == h.getMemorySize()) && ((h.getTriggerType() & (uint32_t)dropEmptyHBFramesTriggerMask) == 0)) {
         return true;
       }
       return false;
@@ -642,6 +649,7 @@ class ConsumerFileRecorder : public Consumer
   int maxFileTF = 0;                  // maximum number of TF to write (in each file)
   int filesMax = 0;                   // maximum number of files to write (for each stream)
   int dropEmptyHBFrames = 0;          // if set, some empty packets are discarded (see logic in code)
+  int dropEmptyHBFramesTriggerMask = 0; // (when using dropEmptyHBFrames = 1) empty HB frames are kept if any bit in RDH TriggerType field matches this pattern. (TriggerType & TriggerMask != 0)
 
   class Packet
   {

src/ConsumerStats.cxx

@@ -194,7 +194,7 @@ class ConsumerStats : public Consumer
         theLog.log(LogInfoOps_(3003), "Last interval (%.2fs): blocksRx=%llu, block rate=%.2lf, block size = %.1lfkB, bytesRx=%llu, rate=%s", deltaT, (unsigned long long)counterBlocksDiff, counterBlocksDiff / deltaT, counterBytesDiff / (1024.0*counterBlocksDiff), (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, released = %llu, release rate=%.2lf Hz, latency=%.3lf s, current TF = %d", (unsigned long long) snapshot.pagesPendingFairMQ.load(), nRfmq, rRfmq, avgTfmq, tfidfmq );
-	  theLog.log(LogInfoOps_(3003), "STFB HBF repacking = %.1lf Hz, copy overhead = %.1lf MB/s", ddHBFRepackedRate, ddBytesCopiedRate);
+	  theLog.log(LogInfoOps_(3003), "STFB HBF repacking = %.1lf Hz, copy overhead = %.1lf MB/s = %.2f%%", ddHBFRepackedRate, ddBytesCopiedRate, ddBytesCopiedRate * 1024.0 * 1024.0 * 100.0 * deltaT / counterBytesDiff);
 	  theLog.log(LogInfoOps_(3003), "STFB memory efficiency = %.1lf %%, data buffered = %.1lf MB, real memory used %.1lf MB",
 	    ddMemoryEfficiency, snapshot.ddPayloadPendingBytes / (1024.0*1024.0), snapshot.ddMemoryPendingBytes / (1024.0*1024.0) );
 	}