Merge pull request #224 from sy-c/master

v2.11

Author: sy-c

doc/configurationParameters.md

@@ -69,6 +69,7 @@ The parameters related to 3rd-party libraries are described here for convenience
 | 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.|
+| consumer-fileRecorder-* | tfMax | int | 0 | Maximum number of timeframes accepted by recorder. If zero (default), no maximum set.|
 | consumer-processor-* | ensurePageOrder | int | 0 | If set, ensures that data pages goes out of the processing pool in same order as input (which is not guaranteed with multithreading otherwise). This option adds latency. |
 | consumer-processor-* | libraryPath | string | | Path to the library file providing the processBlock() function to be used. |
 | consumer-processor-* | numberOfThreads | int | 1 | Number of threads running the processBlock() function in parallel. |
@@ -91,6 +92,7 @@ The parameters related to 3rd-party libraries are described here for convenience
 | consumer-zmq-* | zmqOptions | string |  | Additional ZMQ options, as a comma-separated list of key=value pairs. Possible keys: ZMQ_CONFLATE, ZMQ_IO_THREADS, ZMQ_LINGER, ZMQ_SNDBUF, ZMQ_SNDHWM, ZMQ_SNDTIMEO. |
 | equipment-* | blockAlign | bytes | 2M | Alignment of the beginning of the big memory block from which the pool is created. Pool will start at a multiple of this value. Each page will then begin at a multiple of memoryPoolPageSize from the beginning of big block. |
 | equipment-* | consoleStatsUpdateTime | double | 0 | If set, number of seconds between printing statistics on console. |
+| 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. |
 | equipment-* | dataPagesLogPath | string |  | Path where to save a summary of each data pages generated by equipment. |
 | equipment-* | debugFirstPages | int | 0 | If set, print debug information for first (given number of) data pages readout. |
 | equipment-* | disableOutput | int | 0 | If non-zero, data generated by this equipment is discarded immediately and is not pushed to output fifo of readout thread. Used for testing. |
@@ -116,6 +118,7 @@ The parameters related to 3rd-party libraries are described here for convenience
 | equipment-* | saveErrorPagesPath | string |  | Path where to save data pages with errors (when feature enabled). |
 | equipment-* | stopOnError | int | 0 | If 1, readout will stop automatically on equipment error. |
 | equipment-* | TFperiod | int | 128 | Duration of a timeframe, in number of LHC orbits. |
+| equipment-* | verbose | int | 0 | If set, extra debug messages may be logged. |
 | equipment-cruemulator-* | cruBlockSize | int | 8192 | Size of a RDH block. |
 | equipment-cruemulator-* | cruId | int | 0 | CRU Id, used for CRU Id field in RDH. |
 | equipment-cruemulator-* | dpwId | int | 0 | CRU end-point Id (data path wrapper id), used for DPW Id field in RDH. |

doc/releaseNotes.md

@@ -460,3 +460,11 @@ This file describes the main feature changes for each readout.exe released versi
 
 ## v2.10.4 - 18/05/2022
 - Changed level of "Bypassing RORC firmware compatibility check", retrograded from support warning to developper info.
+
+## v2.11 - 15/06/2022
+- Added orbit counter to stats.
+- Updated configuration parameters:
+  - equipment-*.verbose: adds some extra verbosity.
+  - equipment-*.ctpMode: enables CTP mode, data is discarded until start of run detected in RDH detector field pattern, and after corresponding end of run is detected.
+  - consumer-fileRecorder-*.tfMax: when set, recording will stop after reaching given number of timeframes.
+- o2-readout-rawreader: added option to dump stats on HBF size.

src/ConsumerFileRecorder.cxx

@@ -21,13 +21,15 @@
 class FileHandle
 {
  public:
-  FileHandle(std::string& _path, InfoLogger* _theLog = nullptr, unsigned long long _maxFileSize = 0, int _maxPages = 0)
+  FileHandle(std::string& _path, InfoLogger* _theLog = nullptr, unsigned long long _maxFileSize = 0, int _maxPages = 0, int _maxTF = 0)
   {
     theLog = _theLog;
     path = _path;
     counterBytesTotal = 0;
     maxFileSize = _maxFileSize;
     maxPages = _maxPages;
+    maxTF = _maxTF;
+
     if (theLog != nullptr) {
       theLog->log(LogInfoDevel_(3007), "Opening file for writing: %s", path.c_str());
     }
@@ -62,7 +64,7 @@ class FileHandle
   enum Status { Success = 0,
                 Error = -1,
                 FileLimitsReached = 1 };
-  FileHandle::Status write(void* ptr, size_t size, bool isPage = false, size_t remainingBlockSize = 0)
+  FileHandle::Status write(void* ptr, size_t size, uint64_t TFid, bool isPage = false, size_t remainingBlockSize = 0)
   {
     lastWriteBytes = 0; // reset last bytes written
     if (isFull) {
@@ -88,6 +90,20 @@ class FileHandle
       close();
       return Status::FileLimitsReached;
     }
+    if (TFid != undefinedTimeframeId) {
+      if (TFid != lastTFid) {
+        lastTFid = TFid;
+        counterTF++;
+      }
+    }
+    if ((maxTF) && (counterTF > maxTF)) {
+      if (theLog != nullptr) {
+        theLog->log(LogInfoDevel_(3007), "Maximum number of TF in file reached");
+      }
+      isFull = true;
+      close();
+      return Status::FileLimitsReached;
+    }
     if (fp == NULL) {
       return Status::Error;
     }
@@ -113,6 +129,10 @@ class FileHandle
   unsigned long long maxFileSize = 0;       // max number of bytes to write to file (0=no limit)
   int counterPages = 0;                     // number of pages received so far
   int maxPages = 0;                         // max number of pages accepted by recorder (0=no limit)
+  uint64_t lastTFid = undefinedTimeframeId;      // id of last TF written
+  int counterTF = 0;                        // number of TF received so far
+  int maxTF = 0;                            // max number of timeframes accepted by recorder (0=no limit)
+
   FILE* fp = NULL;                          // handle to file for I/O
   InfoLogger* theLog = nullptr;             // handle to infoLogger for messages
   bool isFull = false;                      // flag set when maximum file size reached
@@ -165,6 +185,14 @@ class ConsumerFileRecorder : public Consumer
       }
     }
 
+    // configuration parameter: | consumer-fileRecorder-* | tfMax | int | 0 | Maximum number of timeframes accepted by recorder. If zero (default), no maximum set.|
+    maxFileTF = 0;
+    if (cfg.getOptionalValue<int>(cfgEntryPoint + ".tfMax", maxFileTF) == 0) {
+      if (maxFileTF) {
+        theLog.log(LogInfoDevel_(3002), "Maximum recording size: %d timeframes", maxFileTF);
+      }
+    }
+
     // configuration parameter: | 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. |
     cfg.getOptionalValue(cfgEntryPoint + ".dataBlockHeaderEnabled", recordWithDataBlockHeader, 0);
     theLog.log(LogInfoDevel_(3002), "Recording internal data block headers = %d", recordWithDataBlockHeader);
@@ -366,7 +394,7 @@ class ConsumerFileRecorder : public Consumer
     }
 
     // create file handle
-    std::shared_ptr<FileHandle> newHandle = std::make_shared<FileHandle>(newFileName, &theLog, maxFileSize, maxFilePages);
+    std::shared_ptr<FileHandle> newHandle = std::make_shared<FileHandle>(newFileName, &theLog, maxFileSize, maxFilePages, maxFileTF);
     if (newHandle == nullptr) {
       return -1;
     }
@@ -429,7 +457,7 @@ class ConsumerFileRecorder : public Consumer
 
     bool countPage = true; // the first write will increment the page counter for this file
 
-    auto writeToFile = [&](void* ptr, size_t size, size_t remainingBlockSize) {
+    auto writeToFile = [&](void* ptr, size_t size, uint64_t TFid, size_t remainingBlockSize) {
       // two attempts, in case file needs to be incremented
       for (int i = 0; i < 2; i++) {
 
@@ -440,7 +468,7 @@ class ConsumerFileRecorder : public Consumer
         }
 
         // try to write
-        FileHandle::Status status = fpUsed->write(ptr, size, countPage, remainingBlockSize);
+        FileHandle::Status status = fpUsed->write(ptr, size, TFid, countPage, remainingBlockSize);
 
         // check if need to move to next file
         if (status == FileHandle::Status::FileLimitsReached) {
@@ -504,14 +532,14 @@ class ConsumerFileRecorder : public Consumer
         // as-is, some fields like data pointer will not be meaningful in file unless corrected.
         // todo: correct them, e.g. replace data pointer by file offset.
         // In particular, incompatible with dropEmptyHBFrames as size changes.
-        writeToFile(&b->getData()->header, (size_t)b->getData()->header.headerSize, (size_t)b->getData()->header.dataSize);
+        writeToFile(&b->getData()->header, (size_t)b->getData()->header.headerSize, b->getData()->header.timeframeId, (size_t)b->getData()->header.dataSize);
         // datablock header does not count as a page, but we account for the payload size for the next write (possibly one full page)
       }
 
       // write payload data
       if (!dropEmptyHBFrames) {
         // by default, we write the full payload data
-        writeToFile(b->getData()->data, (size_t)b->getData()->header.dataSize, 0);
+        writeToFile(b->getData()->data, (size_t)b->getData()->header.dataSize, b->getData()->header.timeframeId, 0);
       } else {
         // we have to check packet by packet and discard empty HBstart/HBstop pairs
         size_t blockSize = b->getData()->header.dataSize;
@@ -545,7 +573,7 @@ class ConsumerFileRecorder : public Consumer
 
           // write previous packet
           if (previousPacket.address != nullptr) {
-            writeToFile(previousPacket.address, previousPacket.size, 0);
+            writeToFile(previousPacket.address, previousPacket.size, previousPacket.timeframeId, 0);
             packetsRecorded++;
             previousPacket.clear();
           }
@@ -568,11 +596,12 @@ class ConsumerFileRecorder : public Consumer
               previousPacket.isCopy = true;
             }
             previousPacket.isEmptyHBStart = true;
+            previousPacket.timeframeId = b->getData()->header.timeframeId;
           } else {
 
             // write packet
             // use offsetNextPacket instead of memorySize for file to be consistent
-            writeToFile(baseAddress + pageOffset, (size_t)h.getOffsetNextPacket(), 0);
+            writeToFile(baseAddress + pageOffset, (size_t)h.getOffsetNextPacket(), b->getData()->header.timeframeId, 0);
             packetsRecorded++;
           }
 
@@ -610,6 +639,7 @@ class ConsumerFileRecorder : public Consumer
   int recordWithDataBlockHeader = 0;  // if set, internal readout headers are included in file
   unsigned long long maxFileSize = 0; // maximum number of bytes to write (in each file)
   int maxFilePages = 0;               // maximum number of pages to write (in each file)
+  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)
 
@@ -620,6 +650,7 @@ class ConsumerFileRecorder : public Consumer
     void* address = nullptr;
     size_t size = 0;
     bool isCopy = false;
+    uint64_t timeframeId = undefinedTimeframeId;
     void clear()
     {
       isEmptyHBStart = false;
@@ -629,6 +660,7 @@ class ConsumerFileRecorder : public Consumer
       address = nullptr;
       size = 0;
       isCopy = false;
+      timeframeId = undefinedTimeframeId;
     }
     Packet() {}
     ~Packet() { clear(); }

src/ConsumerStats.cxx

@@ -189,6 +189,7 @@ class ConsumerStats : public Consumer
 	if (bufferReport.length()) {
 	  theLog.log(LogInfoOps_(3003), "Memory buffers usage: %s", bufferReport.c_str());
 	}
+	//theLog.log(LogInfoOps_(3003), "orbit=0x%X", (unsigned int) gReadoutStats.counters.currentOrbit.load());
       }
     }
 

src/ConsumerZmq.cxx

@@ -105,7 +105,6 @@ class ConsumerZMQ : public Consumer
       }
       zh = zmq_socket(context, ZMQ_PUB);
       if (zh==nullptr) { linerr=__LINE__; zmqerr=zmq_errno(); break; }
-      zmqerr = zmq_bind(zh, cfgAddress.c_str());
       if (zmqerr) { linerr=__LINE__; break; }
       zmqerr = zmq_setsockopt(zh, ZMQ_CONFLATE, &cfg_ZMQ_CONFLATE, sizeof(cfg_ZMQ_CONFLATE));
       if (zmqerr) { linerr=__LINE__; break; }
@@ -117,6 +116,8 @@ class ConsumerZMQ : public Consumer
       if (zmqerr) { linerr=__LINE__; break; }
       zmqerr = zmq_setsockopt(zh, ZMQ_SNDTIMEO, (void*)&cfg_ZMQ_SNDTIMEO, sizeof(cfg_ZMQ_SNDTIMEO));
       if (zmqerr) { linerr=__LINE__; break; }
+      // options (eg CONFLATE) should be set before binding server
+      zmqerr = zmq_bind(zh, cfgAddress.c_str());
       break;
     }
 

src/CounterStats.cxx

@@ -153,3 +153,35 @@ void CounterStats::getHisto(std::vector<double>& x, std::vector<CounterValue>& c
   }
 }
 
+double CounterStats::getStdDev()
+{
+  double sum = 0;
+  CounterValue count = 0;
+  double avg = getAverage();
+  if (histoNbin) {
+    for (unsigned int i = 0; i < histoNbin; i++) {
+      double x;
+      if (histoLogScale) {
+        x = histoVmax * pow(histoStep, histoNbin - 1 - i);
+      } else {
+        if (i == 0) {
+          x = histoVmin;
+        } else if (i == histoNbin - 1) {
+          x = histoVmax;
+        } else {
+          x = histoVmin + (i - 1) * histoStep;
+        }
+      }
+      CounterValue c = histoCounts[i];
+      if (c) {
+        double d = (x-avg);
+        sum += c * d * d;
+        count += c;
+      }
+    }
+  }
+  if (count) {
+    return sqrt(sum/(count-1));
+  }
+  return 0;
+}

src/CounterStats.h

@@ -32,6 +32,7 @@ class CounterStats
   CounterValue get();      // get latest value
   CounterValue getTotal(); // get total of previous values set
   double getAverage();
+  double getStdDev();      // get standard deviation (when histogram enabled)
   CounterValue getMinimum();
   CounterValue getMaximum();
   CounterValue getCount();