Merge pull request #242 from sy-c/master

v2.16.1

Author: sy-c

CMakeLists.txt

@@ -71,7 +71,7 @@ find_package(FairMQ)
 find_package(ZLIB)
 find_package(RDMA)
 find_package(Occ)
-find_package(BookkeepingApiCpp)
+find_package(BookkeepingApi)
 find_package(ZMQ)
 find_package(MySQL)
 #find_package(gperftools)
@@ -116,13 +116,13 @@ else ()
 endif(Occ_FOUND)
 
 # check Bookkeeping
-if (BookkeepingApiCpp_FOUND)
-  get_target_property(BookkeepingApiCpp_INCLUDES BookkeepingApiCpp INTERFACE_INCLUDE_DIRECTORIES)
-#  get_target_property(BookkeepingApiCpp_LIBRARIES BookkeepingApiCpp INTERFACE_LINK_LIBRARIES)
-  message(STATUS "Bookkeeping found")
+if (BookkeepingApi_FOUND)
+  get_target_property(BookkeepingApi_INCLUDES AliceO2::BookkeepingApi INTERFACE_INCLUDE_DIRECTORIES)
+  get_target_property(BookkeepingApi_LIBRARIES AliceO2::BookkeepingApi INTERFACE_LINK_LIBRARIES)
+  message(STATUS "Bookkeeping found: ${BookkeepingApi_INCLUDES} ${BookkeepingApi_LIBRARIES}")
 else ()
   message(STATUS "Bookkeeping not found, corresponding features will be disabled.")
-endif(BookkeepingApiCpp_FOUND)
+endif(BookkeepingApi_FOUND)
 
 # check ZMQ
 if(${ZMQ_FOUND})
@@ -149,7 +149,7 @@ endif()
 
 
 # add flags to enable optional features in Readout, based on available dependencies
-add_compile_definitions($<$<BOOL:${Numa_FOUND}>:WITH_NUMA> $<$<BOOL:${RDMA_FOUND}>:WITH_RDMA> $<$<BOOL:${Configuration_FOUND}>:WITH_CONFIG> $<$<BOOL:${FairMQ_FOUND}>:WITH_FAIRMQ> $<$<BOOL:${Occ_FOUND}>:WITH_OCC> $<$<BOOL:${BookkeepingApiCpp_FOUND}>:WITH_LOGBOOK> $<$<BOOL:${ZMQ_FOUND}>:WITH_ZMQ> $<$<BOOL:${MYSQL_FOUND}>:WITH_DB> $<$<BOOL:${ReadoutCard_FOUND}>:WITH_READOUTCARD> $<$<BOOL:${gperftools_FOUND}>:WITH_GPERFTOOLS>)
+add_compile_definitions($<$<BOOL:${Numa_FOUND}>:WITH_NUMA> $<$<BOOL:${RDMA_FOUND}>:WITH_RDMA> $<$<BOOL:${Configuration_FOUND}>:WITH_CONFIG> $<$<BOOL:${FairMQ_FOUND}>:WITH_FAIRMQ> $<$<BOOL:${Occ_FOUND}>:WITH_OCC> $<$<BOOL:${BookkeepingApi_FOUND}>:WITH_LOGBOOK> $<$<BOOL:${ZMQ_FOUND}>:WITH_ZMQ> $<$<BOOL:${MYSQL_FOUND}>:WITH_DB> $<$<BOOL:${ReadoutCard_FOUND}>:WITH_READOUTCARD> $<$<BOOL:${gperftools_FOUND}>:WITH_GPERFTOOLS>)
 
 # define include directories
 set(READOUT_INCLUDE_DIRS
@@ -161,7 +161,7 @@ set(READOUT_INCLUDE_DIRS
  ${Configuration_INCLUDE_DIRS}
  ${FairMQ_INCLUDE_DIRS}
  ${Occ_INCLUDE_DIRS}
- ${BookkeepingApiCpp_INCLUDES}
+ ${BookkeepingApi_INCLUDES}
 )
 
 # define liraries to be linked
@@ -173,7 +173,7 @@ set(READOUT_LINK_LIBRARIES
   $<$<BOOL:${Configuration_FOUND}>:AliceO2::Configuration>
   $<$<BOOL:${FairMQ_FOUND}>:FairMQ::FairMQ>
   $<$<BOOL:${Occ_FOUND}>:AliceO2::Occ>
-  $<$<BOOL:${BookkeepingApiCpp_FOUND}>:BookkeepingApiCpp>
+  $<$<BOOL:${BookkeepingApi_FOUND}>:AliceO2::BookkeepingApi>
 )
 if(Numa_FOUND)
   list(APPEND READOUT_LINK_LIBRARIES ${Numa_LIBRARIES})
@@ -460,7 +460,7 @@ add_executable(
 )
 
 # a minimal test program to check Bookkeeping library
-if (BookkeepingApiCpp_FOUND)
+if (BookkeepingApi_FOUND)
 add_executable(
         o2-readout-test-lib-bookkeeping
 	${SOURCE_DIR}/testBookkeeping.cxx
@@ -488,7 +488,7 @@ endif()
 if (${ZMQ_FOUND})
   list (APPEND executables o2-readout-test-zmq-client o2-readout-test-zmq-server o2-readout-eventdump)
 endif()
-if (BookkeepingApiCpp_FOUND)
+if (BookkeepingApi_FOUND)
   list (APPEND executables o2-readout-test-lib-bookkeeping)
 endif()
 if (MYSQL_FOUND)

doc/configurationParameters.md

@@ -115,6 +115,7 @@ The parameters related to 3rd-party libraries are described here for convenience
 | equipment-* | rdhCheckDetectorField | int | 0 | If set, the detector field is checked and changes reported. |
 | equipment-* | rdhCheckEnabled | int | 0 | If set, data pages are parsed and RDH headers checked. Errors are reported in logs. |
 | equipment-* | rdhCheckFirstOrbit | int | 1 | If set, it is checked that the first orbit of all equipments and links is the same. If not, run is stopped. |
+| equipment-* | rdhCheckTrigger | int | 0 | If set, the RDH trigger counters are checked for consistency. |
 | equipment-* | rdhDumpEnabled | int | 0 | If set, data pages are parsed and RDH headers summary printed on console. Setting a negative number will print only the first N pages.|
 | equipment-* | rdhDumpErrorEnabled | int | 1 | If set, a log message is printed for each RDH header error found.|
 | equipment-* | rdhDumpFirstInPageEnabled | int | 0 | If set, the first RDH in each data page is logged. Setting a negative number will printit only for the first N pages. |
@@ -132,9 +133,11 @@ The parameters related to 3rd-party libraries are described here for convenience
 | equipment-cruemulator-* | feeId | int | 0 | Front-End Electronics Id, used for FEE Id field in RDH. |
 | equipment-cruemulator-* | HBperiod | int | 1 | Interval between 2 HeartBeat triggers, in number of LHC orbits. |
 | equipment-cruemulator-* | linkId | int | 0 | Id of first link. If numberOfLinks>1, ids will range from linkId to linkId+numberOfLinks-1. |
+| equipment-cruemulator-* | linkThroughput | double | 3.2 | The data throughput of each link, in Gbps. |
 | equipment-cruemulator-* | maxBlocksPerPage | int | 0 | [obsolete- not used]. Maximum number of blocks per page. |
 | equipment-cruemulator-* | numberOfLinks | int | 1 | Number of GBT links simulated by equipment. |
 | equipment-cruemulator-* | PayloadSize | int | 64k | Maximum payload size for each trigger. Actual size is randomized, and then split in a number of (cruBlockSize) packets. |
+| equipment-cruemulator-* | PayloadSizeStdev | double | 0.0 | Standard deviation of randomized PayloadSize (no unit, as a fraction of PayloadSize). |
 | equipment-cruemulator-* | systemId | int | 19 | System Id, used for System Id field in RDH. By default, using the TEST code. |
 | equipment-cruemulator-* | triggerRate | double | 0 | If set, the HB frame rate is limited to given value in Hz (1 HBF per data page). |
 | equipment-dummy-* | eventMaxSize | bytes | 128k | Maximum size of randomly generated event. |

doc/releaseNotes.md

@@ -511,3 +511,12 @@ This file describes the main feature changes for each readout.exe released versi
 - 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.
+
+## v2.16.1 - 16/02/2022
+- Updated configuration parameters:
+  - added equipemnt-*.rdhCheckTrigger: when set, enables some consistency checks on trigger counters, as follows:
+    - Trigger type TF should happen only on orbits % TF period = 0.
+    - First trigger received on each link should be SOC or SOT.
+  - added equipment-cruemulator-*.PayloadSizeStdev: generate payload with random size (gaussian distribution mean=PayloadSize sigma=PayloadSizeStdev).
+  - added equipment-cruemulator-*.linkThroughput: set incoming link data throughput (in Gbps).
+- Minor logging updates.

src/MemoryBankManager.cxx

@@ -13,6 +13,7 @@
 #include "ReadoutUtils.h"
 #include <unistd.h>
 #include <sys/mman.h>
+#include <thread>
 
 #include "readoutInfoLogger.h"
 
@@ -141,7 +142,34 @@ std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize
   #else
     mlock(blockAddress, blockSize);
   #endif
-  bzero(blockAddress, blockSize);
+  const int nMemThreads = 1;
+  if (nMemThreads <= 1) {
+    bzero(blockAddress, blockSize);
+  } else {
+    // parallel
+    std::thread memThreads[nMemThreads];
+    char *ptr = (char *)blockAddress;
+    char *ptrEnd = (char *)blockAddress + blockSize;
+    const size_t blockUnit = 128 * 1024UL * 1024UL; // block unit = 128MB
+    int nThreads = 0;
+    for (int i = 0; i<nMemThreads; i++) {
+      size_t sz = blockSize / nMemThreads;
+      sz = sz + (blockUnit - ((size_t)ptr + sz) % blockUnit); // round up to next block
+      if ((ptr + sz > ptrEnd) || (i + 1 == nMemThreads)) {
+        sz = ptrEnd - ptr;
+      }
+      theLog.log(LogDebugDevel, "Thread %d  - zero %p - %llu", i, ptr, (unsigned long long)sz);
+      memThreads[i] = std::thread(bzero, ptr, sz);
+      ptr += sz;
+      nThreads++;
+      if (ptr >= ptrEnd) {
+        break;
+      }
+    }
+    for (int i = 0; i < nThreads; i++) {
+      memThreads[i].join();
+    }
+  }
   theLog.log(LogInfoDevel, "Zero memory done");
 
   if (numaNode >= 0) {

src/RAWDataHeader.h

@@ -504,6 +504,7 @@ typedef struct _RAWDataHeaderV6 {
 
 } RAWDataHeaderV6;
 
+// as defined in https://gitlab.cern.ch/AliceO2Group/wp6-doc/-/blob/master/rdh/RDHv7.md
 typedef struct _RAWDataHeaderV7 {
   // 32-bits words
 
@@ -640,6 +641,36 @@ typedef struct _RAWDataHeaderV7 {
 
 } RAWDataHeaderV7;
 
+// definition of triggerType RDH field
+typedef struct _RDHTriggerType {
+  union {
+    uint32_t word0 = 0;
+    struct {
+      uint32_t ORBIT : 1;
+      uint32_t HB : 1;
+      uint32_t HBr : 1;
+      uint32_t HC : 1;
+      uint32_t PhT : 1;
+      uint32_t PP : 1;
+      uint32_t Cal : 1;
+      uint32_t SOT : 1;
+      uint32_t EOT : 1;
+      uint32_t SOC : 1;
+      uint32_t EOC : 1;
+      uint32_t TF : 1;
+      uint32_t FErst : 1;
+      uint32_t RT : 1;
+      uint32_t RS : 1;
+      uint32_t spare : 12;
+      uint32_t LHCgap1 : 1;
+      uint32_t LHCgap2 : 1;
+      uint32_t TPCsync : 1;
+      uint32_t TPCrst : 1;
+      uint32_t TOF : 1;
+    };
+  };
+} RDHTriggerType;
+
 using RAWDataHeader = RAWDataHeaderV6;
 
 // expecting 16*32bits = 64 bytes

src/RdhUtils.h

@@ -58,6 +58,7 @@ class RdhHandle
   inline uint32_t getTriggerOrbit() { return (uint32_t)rdhPtr->triggerOrbit; }
   inline uint32_t getTriggerBC() { return (uint32_t)rdhPtr->triggerBC; }
   inline uint32_t getTriggerType() { return (uint32_t)rdhPtr->triggerType; }
+  inline o2::Header::RDHTriggerType getTriggerTypeStruct() { o2::Header::RDHTriggerType tt; tt.word0 = rdhPtr->triggerType; return tt; }
   inline uint16_t getCruId() { return (uint16_t)rdhPtr->cruId; }
   inline uint8_t getEndPointId() { return (uint8_t)rdhPtr->dpwId; }
   inline uint32_t getDetectorField() { return (uint32_t)rdhPtr->detectorField; }

src/ReadoutEquipment.cxx

@@ -143,6 +143,8 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
   cfg.getOptionalValue<int>(cfgEntryPoint + ".rdhCheckFirstOrbit", cfgRdhCheckFirstOrbit);
   // configuration parameter: | equipment-* | rdhCheckDetectorField | int | 0 | If set, the detector field is checked and changes reported. |
   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-* | 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. |
@@ -845,6 +847,14 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
         }
         equipmentLinksData[linkId].firstOrbit = orbitId;
         equipmentLinksData[linkId].firstOrbitIsDefined = 1;
+
+        if (cfgRdhCheckTrigger) {
+          // first data from link should be SOC or SOT trigger type
+          if (! (h.getTriggerTypeStruct().SOC || h.getTriggerTypeStruct().SOT)) {
+            theLog.log(LogErrorSupport_(3241), "Equipment %s : first RDH of link %d is not SOC or SOT, trigger type = 0x%X", name.c_str(), linkId, h.getTriggerTypeStruct().word0);
+          }
+        }
+
       }
     }
   }
@@ -938,6 +948,22 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
           break; // stop checking this page
 	}
       }
+
+      // check trigger counters
+      if (cfgRdhCheckTrigger) {
+        if (h.getTriggerTypeStruct().TF) {
+          // TF boundary should be aligned with TF length
+          if (h.getHbOrbit() % timeframePeriodOrbits) {
+            if (cfgRdhDumpErrorEnabled) {
+              theLog.log(logRdhErrorsToken, "Equipment %d Link %d RDH %d @ 0x%X : TriggerType TF bit set, but orbit 0x%08X not aligned with TF period = %d", id, (int)blockHeader.linkId, rdhIndexInPage, (unsigned int)pageOffset, (int)h.getTriggerOrbit(), (int)timeframePeriodOrbits);
+              isPageError = 1;
+            }
+            statsRdhCheckStreamErr++;
+            break; // stop checking this page
+          }
+        }
+      }
+
       /*
       // check packetCounter is contiguous
       if (cfgRdhCheckPacketCounterContiguous) {

src/ReadoutEquipment.h

@@ -176,6 +176,7 @@ class ReadoutEquipment
   int cfgRdhUseFirstInPageEnabled = 0; // flag to enable reading of first RDH in page to populate readout headers
   int cfgRdhDumpFirstInPageEnabled = 0;// flag to enable RDH dump of first header in page
   int cfgRdhCheckFirstOrbit = 1;       // flag to enable RDH check of first orbit is the same in all equipments
+  int cfgRdhCheckTrigger = 0;          // flag to enable RDH check of trigger counters
   //int cfgRdhCheckPacketCounterContiguous = 1; // flag to enable checking if RDH packetCounter value contiguous (done link-by-link)
   int cfgRdhCheckDetectorField = 0; // flag to enable checking for changes in detector field
   double cfgTfRateLimit = 0;           // TF rate limit, to throttle data readout

src/ReadoutEquipmentCruEmulator.cxx

@@ -12,6 +12,8 @@
 #include <Common/Fifo.h>
 #include <Common/Timer.h>
 #include <stdlib.h>
+#include <random>
+#include <cmath>
 
 #include "RAWDataHeader.h"
 #include "ReadoutEquipment.h"
@@ -54,6 +56,11 @@ class ReadoutEquipmentCruEmulator : public ReadoutEquipment
 
   double cfgEmptyHbRatio = 0.0;   // amount of empty HB frames
   int cfgPayloadSize = 64 * 1024; // maximum payload size, randomized
+  double cfgPayloadSizeStdev = 0.0; // standard deviation of randomized payload size
+
+  std::random_device payloadRd{};
+  std::mt19937 payloadGen{payloadRd()};
+  std::normal_distribution<double> payloadDistrib{};
 
   double cfgTriggerRate = 0.0; // if set, generate blocks at given rate instead of continuously
   unsigned long long nBlocksPerLink = 0; // number of blocks sent
@@ -93,7 +100,9 @@ ReadoutEquipmentCruEmulator::ReadoutEquipmentCruEmulator(ConfigFile& cfg, std::s
   // configuration parameter: | equipment-cruemulator-* | HBperiod | int | 1 | Interval between 2 HeartBeat triggers, in number of LHC orbits. |
   // configuration parameter: | equipment-cruemulator-* | EmptyHbRatio | double | 0 | Fraction of empty HBframes, to simulate triggered detectors. |
   // configuration parameter: | equipment-cruemulator-* | PayloadSize | int | 64k | Maximum payload size for each trigger. Actual size is randomized, and then split in a number of (cruBlockSize) packets. |
+  // configuration parameter: | equipment-cruemulator-* | PayloadSizeStdev | double | 0.0 | Standard deviation of randomized PayloadSize (no unit, as a fraction of PayloadSize). |
   // configuration parameter: | equipment-cruemulator-* | triggerRate | double | 0 | If set, the HB frame rate is limited to given value in Hz (1 HBF per data page). |
+  // configuration parameter: | equipment-cruemulator-* | linkThroughput | double | 3.2 | The data throughput of each link, in Gbps. |
   cfg.getOptionalValue<int>(cfgEntryPoint + ".maxBlocksPerPage", cfgMaxBlocksPerPage, (int)0);
   cfg.getOptionalValue<int>(cfgEntryPoint + ".cruBlockSize", cruBlockSize, (int)8192);
   cfg.getOptionalValue<int>(cfgEntryPoint + ".numberOfLinks", cfgNumberOfLinks, (int)1);
@@ -105,10 +114,12 @@ ReadoutEquipmentCruEmulator::ReadoutEquipmentCruEmulator(ConfigFile& cfg, std::s
   cfg.getOptionalValue<int>(cfgEntryPoint + ".HBperiod", cfgHBperiod);
   cfg.getOptionalValue<double>(cfgEntryPoint + ".EmptyHbRatio", cfgEmptyHbRatio);
   cfg.getOptionalValue<int>(cfgEntryPoint + ".PayloadSize", cfgPayloadSize);
+  cfg.getOptionalValue<double>(cfgEntryPoint + ".PayloadSizeStdev", cfgPayloadSizeStdev);
   cfg.getOptionalValue<double>(cfgEntryPoint + ".triggerRate", cfgTriggerRate);
+  cfg.getOptionalValue<double>(cfgEntryPoint + ".linkThroughput", cfgGbtLinkThroughput);
 
   // log config summary
-  theLog.log(LogInfoDevel_(3002), "Equipment %s: maxBlocksPerPage=%d cruBlockSize=%d numberOfLinks=%d systemId=%d cruId=%d dpwId=%d feeId=%d linkId=%d HBperiod=%d EmptyHbRatio=%f PayloadSize=%d TriggerRate=%f", name.c_str(), cfgMaxBlocksPerPage, cruBlockSize, cfgNumberOfLinks, cfgSystemId, cfgCruId, cfgDpwId, cfgFeeId, cfgLinkId, cfgHBperiod, cfgEmptyHbRatio, cfgPayloadSize, cfgTriggerRate);
+  theLog.log(LogInfoDevel_(3002), "Equipment %s: maxBlocksPerPage=%d cruBlockSize=%d numberOfLinks=%d systemId=%d cruId=%d dpwId=%d feeId=%d linkId=%d HBperiod=%d EmptyHbRatio=%f PayloadSize=%d PayloadSizeStdev=%f TriggerRate=%f linkThroughput=%f", name.c_str(), cfgMaxBlocksPerPage, cruBlockSize, cfgNumberOfLinks, cfgSystemId, cfgCruId, cfgDpwId, cfgFeeId, cfgLinkId, cfgHBperiod, cfgEmptyHbRatio, cfgPayloadSize, cfgPayloadSizeStdev, cfgTriggerRate, cfgGbtLinkThroughput);
 
   // initialize array of pending blocks (to be filled with data)
   pendingBlocks.resize(cfgNumberOfLinks);
@@ -120,8 +131,11 @@ ReadoutEquipmentCruEmulator::ReadoutEquipmentCruEmulator(ConfigFile& cfg, std::s
   }
 
   // init parameters
-  bcStep = (int)(LHCBCRate * ((cruBlockSize - sizeof(o2::Header::RAWDataHeader)) * 1.0 / (cfgGbtLinkThroughput * 1024 * 1024 * 1024 / 8)));
+  bcStep = (int)ceil(LHCBCRate * (double)(cfgPayloadSize + sizeof(o2::Header::RAWDataHeader) * ceil(cfgPayloadSize * 1.0 / (cruBlockSize - sizeof(o2::Header::RAWDataHeader)))) * 8 / (cfgGbtLinkThroughput * 1000 * 1000 * 1000));
   theLog.log(LogInfoDevel_(3002), "Equipment %s: using block rate = %d BC", name.c_str(), bcStep);
+
+  // random payload distribution
+  payloadDistrib = std::normal_distribution<> (cfgPayloadSize, cfgPayloadSizeStdev * cfgPayloadSize);
 }
 
 ReadoutEquipmentCruEmulator::~ReadoutEquipmentCruEmulator() {}
@@ -243,7 +257,10 @@ Thread::CallbackResult ReadoutEquipmentCruEmulator::prepareBlocks()
         } else {
           // HB with random payload size
           ls.isEmpty = 0;
-          ls.payloadBytesLeft = cfgPayloadSize * (rand() * 1.0 / RAND_MAX);
+          ls.payloadBytesLeft = std::round(payloadDistrib(payloadGen));
+          if (ls.payloadBytesLeft < 0 ) {
+            ls.payloadBytesLeft = 0;
+          }
         }
 
       } else {

src/ReadoutEquipmentZmq.cxx

@@ -358,7 +358,7 @@ int ReadoutEquipmentZmq::tfClientCallback(void* msg, int msgSize)
     tfUpdateTime = time(NULL);
     if (tfUpdateTimeWarning) {
       tfUpdateTimeWarning = 0;
-      theLog.log(LogInfoSupport_(3236), "New TF id received from TF server");
+      theLog.log(LogInfoDevel_(3236), "New TF id received from TF server");
     }
     if (maxTf == 0) {
       // that's the first TF