Merge pull request #260 from sy-c/master

v2.21.0

Author: sy-c

doc/configurationParameters.md

@@ -130,6 +130,7 @@ The parameters related to 3rd-party libraries are described here for convenience
 | consumer-zmq-* | maxRate | int| 0 | Maximum number of pages to publish per second. The associated memory copy has an impact on cpu load, so this should be limited when one does not use all the data (eg for eventDump). |
 | consumer-zmq-* | pagesPerBurst | int | 1 | Number of consecutive pages guaranteed to be part of each publish sequence. The maxRate limit is checked at the end of each burst. |
 | 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-* | autoTimeframeId | int | 0 | When set, timeframe IDs are generated incrementally instead of being computed from trigger orbit counters. Useful to replay files with unordered / gap between TF. BC still used to detect boundaries between TFs. |
 | 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. |
@@ -210,8 +211,10 @@ The parameters related to 3rd-party libraries are described here for convenience
 | readout | maxMsgError | int | 0 | If non-zero, maximum number of error messages allowed while running. Readout stops when threshold is reached. |
 | readout | maxMsgWarning | int | 0 | If non-zero, maximum number of error messages allowed while running. Readout stops when threshold is reached. |
 | readout | memoryPoolStatsEnabled | int | 0 | Global debugging flag to enable statistics on memory pool usage (printed to stdout when pool released). |
+| readout | numberOfRuns | int | 1 | In standalone mode, number of runs to execute (ie START/STOP cycles). |
 | readout | rate | double | -1 | Data rate limit, per equipment, in Hertz. -1 for unlimited. |
 | readout | tfRateLimit | double | 0 | When set, the output is limited to a given timeframe rate. |
+| readout | tfRateLimitMode | int | 0 | Defines the source for TF rate limit: 0 = use TF id, 1 = use number of TF. |
 | readout | timeframeServerUrl | string | | The address to be used to publish current timeframe, e.g. to be used as reference clock for other readout instances. |
 | readout | timeStart | string | | In standalone mode, time at which to execute start. If not set, immediately. |
 | readout | timeStop | string | | In standalone mode, time at which to execute stop. If not set, on int/term/quit signal. |

doc/releaseNotes.md

@@ -571,5 +571,9 @@ This file describes the main feature changes for each readout.exe released versi
 - Added consistency check of orbit vs timestamp when large gas in TF ids detected.
 - Updated configuration parameters documentation.
 
-## next version
+## v2.21.0 - 03/08/2023
 - When running from the command line, the environment variable O2_RUN can be used to set the run number. It is set to 0 by default, i.e. undefined run number.
+- Updated configuration parameters:
+  - added readout.numberOfRuns: in standalone mode, number of START/STOP cycles to execute (used for testing).
+  - added readout.tfRateLimitMode: can be set to 1 to use number of TF instead of computed TF id for rate throttling. Useful when replaying files with jumps in TF ids. (not needed with autoTimeframeId)
+  - added equipment.autoTimeframeId: to force incremental timeframe IDs. Useful when replaying files with jumps in TF ids. BC still used to detect boundaries between TFs.

src/DataBlockAggregator.cxx

@@ -292,16 +292,14 @@ int DataBlockSlicer::appendBlock(DataBlockContainerReference const& block, doubl
     }
   }
 
-  // theLog.log(LogDebugTrace, "slicer %p append block eq %d link %d for tf %d",
-  //   this,(int)sourceId.equipmentId,(int)sourceId.linkId,(int)tfId);
+  // theLog.log(LogDebugTrace, "slicer %p append block eq %d link %d for tf %d", this,(int)sourceId.equipmentId,(int)sourceId.linkId,(int)tfId);
   PartialSlice& s = partialSlices[sourceId];
 
   if (s.currentDataSet != nullptr) {
-    // theLog.log(LogDebugTrace, "slice size = %d chunks",partialSlices[linkId].currentDataSet->size()); if ((partialSlices[linkId].tfId!=tfId)||(partialSlices[linkId].currentDataSet->size()>2))
-    // {
+    // theLog.log(LogDebugTrace, "slice size = %d chunks",(int)s.currentDataSet->size());
     if ((s.tfId != tfId) || (tfId == undefinedTimeframeId)) {
       // the current slice is complete
-      // theLog.log(LogDebugTrace, "slicer %p TF %d eq %d link %d is complete (%d blocks)",this, (int)s.tfId,(int)sourceId.equipmentId,(int)sourceId.linkId,s.currentDataSet->size());
+      // theLog.log(LogDebugTrace, "slicer %p TF %d eq %d link %d is complete (%d blocks)",this, (int)s.tfId,(int)sourceId.equipmentId,(int)sourceId.linkId,(int)s.currentDataSet->size());
       slices.push(s.currentDataSet);
       s.currentDataSet = nullptr;
     }
@@ -316,7 +314,7 @@ int DataBlockSlicer::appendBlock(DataBlockContainerReference const& block, doubl
   s.currentDataSet->push_back(block);
   s.tfId = tfId;
   s.lastUpdateTime = timestamp;
-  // printf(" %d,%d -> %d blocks\n",s.sourceId.equipmentId,s.sourceId.linkId,s.currentDataSet->size());
+  // printf(" %d,%d -> %d blocks\n",(int)sourceId.equipmentId,(int)sourceId.linkId,(int)s.currentDataSet->size());
   return s.currentDataSet->size();
 }
 
@@ -340,6 +338,7 @@ DataSetReference DataBlockSlicer::getSlice(bool includeIncomplete)
     bcv = slices.front();
     slices.pop();
   }
+  // printf("getSlice -> %p\n", bcv.get());
   return bcv;
 }
 

src/ReadoutEquipment.cxx

@@ -166,6 +166,9 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
     cfg.getOptionalValue<int>(cfgEntryPoint + ".TFperiod", cfgTFperiod);
     timeframePeriodOrbits = cfgTFperiod;
 
+    // configuration parameter: | equipment-* | autoTimeframeId | int | 0 | When set, timeframe IDs are generated incrementally instead of being computed from trigger orbit counters. Useful to replay files with unordered / gap between TF. BC still used to detect boundaries between TFs. |
+    cfg.getOptionalValue<int>(cfgEntryPoint + ".autoTimeframeId", cfgAutoTimeframeId, 0);
+
     if (!cfgRdhUseFirstInPageEnabled) {
       usingSoftwareClock = true; // if RDH disabled, use internal clock for TF id
     }
@@ -174,7 +177,11 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
       timeframeRate = LHCOrbitRate * 1.0 / timeframePeriodOrbits; // timeframe rate, in Hz
       theLog.log(LogInfoDevel_(3002), "Timeframe IDs generated by software, %.2lf Hz", timeframeRate);
     } else {
-      theLog.log(LogInfoDevel_(3002), "Timeframe IDs generated from RDH trigger counters");
+      if (cfgAutoTimeframeId) {
+        theLog.log(LogInfoDevel_(3002), "Timeframe IDs generated incrementally with boundaries detected from RDH trigger counters");
+      } else {
+        theLog.log(LogInfoDevel_(3002), "Timeframe IDs generated from RDH trigger counters");
+      }
     }
   }
 
@@ -465,6 +472,15 @@ Thread::CallbackResult ReadoutEquipment::threadCallback(void* arg)
 	  if (nextBlock->getData()->header.timeframeId == undefinedTimeframeId) {
             nextBlock->getData()->header.timeframeId = ptr->getCurrentTimeframe();
 	  }
+
+          if (ptr->cfgAutoTimeframeId) {
+            uint64_t tfId = nextBlock->getData()->header.timeframeId;
+            if (tfId != ptr->autoTimeframeIdLatestFromBC) {
+              ptr->autoTimeframeIdLatestFromBC = tfId;
+              ptr->autoTimeframeIdCounter++;
+            }
+            nextBlock->getData()->header.timeframeId = ptr->autoTimeframeIdCounter;
+          }
 	}
 
 	// tag data with run number
@@ -614,6 +630,9 @@ void ReadoutEquipment::initCounters()
 
   statsNumberOfTimeframes = 0;
 
+  autoTimeframeIdLatestFromBC = undefinedTimeframeId;
+  autoTimeframeIdCounter = 0;
+
   // reset timeframe clock
   currentTimeframe = undefinedTimeframeId;
   lastTimeframe = undefinedTimeframeId;

src/ReadoutEquipment.h

@@ -184,6 +184,9 @@ class ReadoutEquipment
   int cfgDisableTimeframes = 0;        // When set, all TF features disabled
   RateRegulator TFregulator;           // clock counter for TF rate checks
   DataBlockContainerReference throttlePendingBlock; // in case TF rate limit was reached, a block may be set aside for later (when it belongs to next TF)
+  int cfgAutoTimeframeId = 0; // when set, TFids are generated incrementally instead of taken from RDH BC.
+  uint64_t autoTimeframeIdLatestFromBC = undefinedTimeframeId; // latest TFid (computed from BC)
+  uint64_t autoTimeframeIdCounter = 0; // TFid counter for sequence
 
   bool isRdhEquipment = false; // to be set true for RDH equipments
 

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.20.0"
+#define READOUT_VERSION "2.21.0"
 

src/mainReadout.cxx

@@ -334,6 +334,7 @@ class Readout
   bool standaloneMode = false; // flag set when readout running in standalone mode (auto state machines)
   int cfgTimeStart = 0;        // time at which START should be executed in standalone mode
   int cfgTimeStop = 0;         // time at which STOP should be executed in standalone mode
+  int cfgNumberOfRuns = 1;           // number of START/STOP loops to execute
 
  private:
   int _init(int argc, char* argv[]);
@@ -357,6 +358,7 @@ class Readout
   double cfgAggregatorSliceTimeout;
   double cfgAggregatorStfTimeout;
   double cfgTfRateLimit;
+  int cfgTfRateLimitMode;
   int cfgLogbookEnabled;
   std::string cfgLogbookUrl;
   std::string cfgLogbookApiToken;
@@ -406,7 +408,8 @@ class Readout
   void publishLogbookStats();          // publish current readout counters to logbook
   AliceO2::Common::Timer logbookTimer; // timer to handle readout logbook publish interval
 
-  uint64_t maxTimeframeId;
+  uint64_t currentTimeframeId = undefinedTimeframeId;
+  uint64_t countTimeframeId;
 
 #ifdef WITH_ZMQ
   std::unique_ptr<ZmqServer> tfServer;
@@ -902,6 +905,9 @@ int Readout::_configure(const boost::property_tree::ptree& properties)
 
     // configuration parameter: | readout | timeStop | string | | In standalone mode, time at which to execute stop. If not set, on int/term/quit signal. |
     scanTime("readout.timeStop", cfgTimeStop);
+
+    // configuration parameter: | readout | numberOfRuns | int | 1 | In standalone mode, number of runs to execute (ie START/STOP cycles). |
+    cfg.getOptionalValue<int>("readout.numberOfRuns", cfgNumberOfRuns);
   }
 
   cfgMaxMsgError = 0;
@@ -934,6 +940,9 @@ int Readout::_configure(const boost::property_tree::ptree& properties)
   // configuration parameter: | readout | tfRateLimit | double | 0 | When set, the output is limited to a given timeframe rate. |
   cfgTfRateLimit = 0;
   cfg.getOptionalValue<double>("readout.tfRateLimit", cfgTfRateLimit);
+  // configuration parameter: | readout | tfRateLimitMode | int | 0 | Defines the source for TF rate limit: 0 = use TF id, 1 = use number of TF. |
+  cfgTfRateLimitMode = 0;
+  cfg.getOptionalValue<int>("readout.tfRateLimitMode", cfgTfRateLimitMode);
 
   // configuration parameter: | readout | disableTimeframes | int | 0 | When set, all timeframe related features are disabled (this may supersede other config parameters). |
   cfgDisableTimeframes = 0;
@@ -1386,7 +1395,8 @@ int Readout::_start()
   #endif
   publishLogbookStats();
   logbookTimer.reset(cfgLogbookUpdateInterval * 1000000);
-  maxTimeframeId = 0;
+  currentTimeframeId = undefinedTimeframeId;
+  countTimeframeId = 0;
 
   // execute custom command
   executeCustomCommand("preSTART");
@@ -1494,24 +1504,30 @@ void Readout::loopRunning()
         if (bc->size() > 0) {
           if (bc->at(0)->getData() != nullptr) {
             uint64_t newTimeframeId = bc->at(0)->getData()->header.timeframeId;
-            // are we complying with maximum TF rate ?
-            if (cfgTfRateLimit > 0) {
-              if (newTimeframeId > floor(startTimer.getTime() * cfgTfRateLimit) + 1) {
-                usleep(1000);
-                continue;
+            if (newTimeframeId != currentTimeframeId) {
+              // beginning of new TF detected: are we complying with maximum TF rate ?
+              if (cfgTfRateLimit > 0) {
+                uint64_t maxTimeframes = floor(startTimer.getTime() * cfgTfRateLimit) + 1; // number of TF allowed at this point
+                // mode 0: compare with TFid
+                // mode 1: use number of TF instead of computed TF id. Useful when replaying files with jumps in TF ids.
+                if (    ((cfgTfRateLimitMode == 0) && (newTimeframeId > maxTimeframes))
+                     || ((cfgTfRateLimitMode == 1) && (countTimeframeId >= maxTimeframes)) ) {
+                  usleep(1000);
+                  continue;
+                }
               }
-            }
-            if (newTimeframeId > maxTimeframeId) {
-              maxTimeframeId = newTimeframeId;
+              countTimeframeId++;
+              currentTimeframeId = newTimeframeId;
 #ifdef WITH_ZMQ
               if (tfServer) {
-                tfServer->publish(&maxTimeframeId, sizeof(maxTimeframeId));
+                tfServer->publish(&currentTimeframeId, sizeof(currentTimeframeId));
               }
 #endif
               gReadoutStats.counters.numberOfSubtimeframes++;
 	      gReadoutStats.counters.currentOrbit =  bc->at(0)->getData()->header.timeframeOrbitFirst;
 	      gReadoutStats.counters.notify++;
             }
+            // printf("Pushing TF #%d = %d\n", (int)countTimeframeId, (int)newTimeframeId);
           }
           for(auto const& b : *bc) {
             updatePageStateFromDataBlockContainerReference(b, MemoryPage::PageState::InConsumer);
@@ -2253,7 +2269,8 @@ int main(int argc, char* argv[])
       return err;
     }
 
-    int nloop = 3; // number of start/stop loop to execute
+    int nloop = theReadout->cfgNumberOfRuns; // number of start/stop loop to execute
+    theLog.log("Will execute %d START/STOP cycle", nloop);
 
     auto logTimeGuard = [&](const std::string command, int t) {
       if (t) {

src/readoutConfigEditor.tcl

@@ -76,6 +76,7 @@ set configurationParametersDescriptor {
 | consumer-zmq-* | maxRate | int| 0 | Maximum number of pages to publish per second. The associated memory copy has an impact on cpu load, so this should be limited when one does not use all the data (eg for eventDump). |
 | consumer-zmq-* | pagesPerBurst | int | 1 | Number of consecutive pages guaranteed to be part of each publish sequence. The maxRate limit is checked at the end of each burst. |
 | 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-* | autoTimeframeId | int | 0 | When set, timeframe IDs are generated incrementally instead of being computed from trigger orbit counters. Useful to replay files with unordered / gap between TF. BC still used to detect boundaries between TFs. |
 | 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. |
@@ -156,8 +157,10 @@ set configurationParametersDescriptor {
 | readout | maxMsgError | int | 0 | If non-zero, maximum number of error messages allowed while running. Readout stops when threshold is reached. |
 | readout | maxMsgWarning | int | 0 | If non-zero, maximum number of error messages allowed while running. Readout stops when threshold is reached. |
 | readout | memoryPoolStatsEnabled | int | 0 | Global debugging flag to enable statistics on memory pool usage (printed to stdout when pool released). |
+| readout | numberOfRuns | int | 1 | In standalone mode, number of runs to execute (ie START/STOP cycles). |
 | readout | rate | double | -1 | Data rate limit, per equipment, in Hertz. -1 for unlimited. |
 | readout | tfRateLimit | double | 0 | When set, the output is limited to a given timeframe rate. |
+| readout | tfRateLimitMode | int | 0 | Defines the source for TF rate limit: 0 = use TF id, 1 = use number of TF. |
 | readout | timeframeServerUrl | string | | The address to be used to publish current timeframe, e.g. to be used as reference clock for other readout instances. |
 | readout | timeStart | string | | In standalone mode, time at which to execute start. If not set, immediately. |
 | readout | timeStop | string | | In standalone mode, time at which to execute stop. If not set, on int/term/quit signal. |