TF rate limit fix

Author: sy-c

src/ReadoutEquipmentPlayer.cxx

@@ -55,7 +55,8 @@ class ReadoutEquipmentPlayer : public ReadoutEquipment
 
   uint32_t orbitOffset = 0; // to be applied to orbit after 1st loop
   int cfgUpdateOrbits = 1; // when set, all RDHs are modified to update orbit, according to orbitOffset
-
+  int cfgAutoTimeframeId = 0; // when set, TFids are generated incrementally instead of taken from RDH BC.
+  
   void copyFileDataToPage(void* page); // fill given page with file data according to current settings
 };
 
@@ -102,6 +103,8 @@ ReadoutEquipmentPlayer::ReadoutEquipmentPlayer(ConfigFile& cfg, std::string cfgE
   cfg.getOptionalValue<int>(cfgEntryPoint + ".autoChunkLoop", autoChunkLoop, 0);
   // configuration parameter: | equipment-player-* | updateOrbits | int | 1 | When set, trigger orbit counters in all RDH are modified for iterations after the first one (in file loop replay mode), so that they keep increasing. |
   cfg.getOptionalValue<int>(cfgEntryPoint + ".updateOrbits", cfgUpdateOrbits, 1);
+  // configuration parameter: | equipment-player-* | 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);
 
   // log config summary
   theLog.log(LogInfoDevel_(3002), "Equipment %s: using data source file=%s preLoad=%d fillPage=%d autoChunk=%d autoChunkLoop=%d updateOrbits=%d", name.c_str(), filePath.c_str(), preLoad, fillPage, autoChunk, autoChunkLoop, cfgUpdateOrbits);

src/mainReadout.cxx

@@ -408,7 +408,7 @@ 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
@@ -1395,7 +1395,7 @@ int Readout::_start()
   #endif
   publishLogbookStats();
   logbookTimer.reset(cfgLogbookUpdateInterval * 1000000);
-  maxTimeframeId = 0;
+  currentTimeframeId = undefinedTimeframeId;
   countTimeframeId = 0;
 
   // execute custom command
@@ -1504,29 +1504,30 @@ void Readout::loopRunning()
         if (bc->size() > 0) {
           if (bc->at(0)->getData() != nullptr) {
             uint64_t newTimeframeId = bc->at(0)->getData()->header.timeframeId;
-            if (cfgTfRateLimitMode == 1) {
-              // use number of TF instead of computed TF id. Useful when replaying files with jumps in TF ids.
-              newTimeframeId = countTimeframeId + 1;
-            }
-            // 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;
+                }
               }
-            }
-            countTimeframeId++;
-            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);