Merge pull request #248 from sy-c/master

v2.18.1

Author: sy-c

doc/releaseNotes.md

@@ -540,3 +540,7 @@ This file describes the main feature changes for each readout.exe released versi
 - Minor internal changes (not for users):
   - equipment-player: stop generating data at beginning of EOR, for faster flushing.
   - consumer.FairMQchannel: using thread mode by default for DD formatting (still 1 thread by default, but better data checks than without threading mode). Using non-blocking FMQ send for faster handling of EOR if no process on the other side.
+
+## v2.18.1 - 23/03/2023
+- Optimized EOR sequence in case FMQ stuck, as per timeout readout.flushConsumerTimeout definition. Pending data are discarded.
+- Moved FMQ repack stats logs at STOP instead of RESET.

src/Consumer.h

@@ -32,6 +32,7 @@ class Consumer
   // Function called just before starting data taking. Data will soon start to flow in.
   virtual int start()
   {
+    isRunning = 1;
     totalPushSuccess = 0;
     totalPushError = 0;
     totalBlocksFiltered = 0;
@@ -42,6 +43,7 @@ class Consumer
   // Function called just after stopping data taking, after the last call to pushData(). Not called before input FIFO empty.
   virtual int stop()
   {
+    isRunning = 0;
     theLog.log(LogInfoDevel_(3003), "Push statistics for %s: %llu err / %llu total (DataSets), %llu/%llu filtered (DataBlocks)",
       this->name.c_str(), totalPushError.load(), totalPushError.load() + totalPushSuccess.load(), totalBlocksFiltered.load(), totalBlocksUnfiltered.load() + totalBlocksFiltered.load());
     return 0;
@@ -65,6 +67,7 @@ class Consumer
   // if link in list of included filters, or included filters list empty: 1
   // empty datablocks are always excluded
   bool isDataBlockFilterOk(const DataBlock&);
+  bool isRunning = 0;                         // flag to indicate running state
 
  private:
   bool filterLinksEnabled = 0;                // when set, defines a filter link

src/ConsumerFMQchannel.cxx

@@ -200,11 +200,13 @@ class ConsumerFMQchannel : public Consumer
     std::unique_ptr<AliceO2::Common::Fifo<wThreadInput>> input;
     std::unique_ptr<AliceO2::Common::Fifo<wThreadOutput>> output;
     std::unique_ptr<std::thread> thread;
+    bool isRunning;
   };
   std::vector<wThread> wThreads;
   int wThreadShutdown = 0;
   const int wThreadSleepTime = 1000; // sleep time in microseconds.   
   std::unique_ptr<std::thread> senderThread; // this one empties the output FIFOs of the wThreads
+  bool senderThreadIsRunning;
   int wThreadIxWrite = 0; // push data round-robin in wThreads
   int wThreadIxRead = 0; // read data round-robin in wThreads
   void cleanupThreads() {
@@ -224,11 +226,15 @@ class ConsumerFMQchannel : public Consumer
       nwThreads = 0;
     }
   }
+  std::atomic<uint64_t> nTFdiscardedEOR = 0;
 
   int processForDataDistribution(DataSetReference& bc);
 
  public:
 
+  int start();
+  int stop();
+
   ConsumerFMQchannel(ConfigFile& cfg, std::string cfgEntryPoint) : Consumer(cfg, cfgEntryPoint)
   {
 
@@ -445,16 +451,6 @@ class ConsumerFMQchannel : public Consumer
     // stop threads
     cleanupThreads();
 
-    // log memory pool statistics
-    if (mp!=nullptr) {
-      theLog.log(LogInfoDevel_(3003), "Consumer %s - memory pool statistics ... %s", name.c_str(), mp->getStats().c_str());
-      theLog.log(LogInfoDevel_(3003), "Consumer %s - STFB repacking statistics ... number: %" PRIu64 " average page size: %" PRIu64 " max page size: %" PRIu64 " repacked/received = %" PRIu64 "/%" PRIu64 " = %.1f%%", name.c_str(), repackSizeStats.getCount(), (uint64_t)repackSizeStats.getAverage(), repackSizeStats.getMaximum(), nPagesUsedForRepack, nPagesUsedInput, nPagesUsedForRepack * 100.0 / nPagesUsedInput);
-    }
-    
-    if (TFdropped) {
-      theLog.log(LogInfoSupport_(3235), "Consumer %s - %llu incomplete TF dropped", name.c_str(), (unsigned long long)TFdropped);
-    }
-
     // release in reverse order
     mp = nullptr;
     memoryBuffer = nullptr; // warning: data range may still be referenced in memory bank manager
@@ -643,6 +639,19 @@ class ConsumerFMQchannel : public Consumer
        break;
      }
 
+     if (!isRunning) {
+       // when not running, empty incoming buffer and get ready to start
+       wThreadInput tf;
+       while ( wThreads[thIx].input->pop(tf) == 0) {
+         nTFdiscardedEOR++;
+       }
+       pushCount = 0;
+       wThreads[thIx].isRunning = 0;
+       usleep(wThreadSleepTime);
+       continue;
+     }
+     wThreads[thIx].isRunning = 1;
+
      // wait that there is a slot in outgoing FIFO
      if (wThreads[thIx].output->isFull()) {
        usleep(wThreadSleepTime);
@@ -716,6 +725,22 @@ class ConsumerFMQchannel : public Consumer
        break;
      }
 
+     if (!isRunning) {
+       // when not running, empty output buffer and get ready to start
+       for (thIx = 0; thIx < nwThreads; thIx++) {
+         wThreadOutput msglist;
+         while ( wThreads[thIx].output->pop(msglist) == 0) {
+           nTFdiscardedEOR++;
+         }
+       }
+       thIx = 0;
+       lastTimeframeId = undefinedTimeframeId;
+       senderThreadIsRunning = 0;
+       usleep(wThreadSleepTime);
+       continue;
+     }
+     senderThreadIsRunning = 1;
+
      // get a TF from FIFO
      wThreadOutput msglist;
      if (wThreads[thIx].output->pop(msglist) != 0) {
@@ -748,9 +773,12 @@ class ConsumerFMQchannel : public Consumer
        if (DDsendMessage(msg)) {
          // sending failed
 	 isError = 1;
-      }
+       }
      }
      if (isError) {
+       if (!isRunning) {
+         nTFdiscardedEOR++; // account for this one at EOR flush
+       }
        totalPushError++;
      }
 
@@ -1198,16 +1226,19 @@ int ConsumerFMQchannel::DDformatMessage(DataSetReference &bc, DDMessage &ddm) {
 int ConsumerFMQchannel::DDsendMessage(DDMessage &ddm) {
   // send the messages
   int err;
-  while (!wThreadShutdown) {
-    err = sendingChannel->Send(ddm.messagesToSend, 500);
+  while ((!wThreadShutdown) && (isRunning)) {
+    err = sendingChannel->Send(ddm.messagesToSend, 1 + wThreadSleepTime / 100);
     if (err>=0) break;
   }
   if ( err >= 0) {
     gReadoutStats.counters.bytesFairMQ += ddm.subTimeframeTotalSize;
     gReadoutStats.counters.timeframeIdFairMQ = ddm.stfHeader->timeframeId;
     gReadoutStats.counters.notify++;
   } else {
-    theLog.log(LogErrorSupport_(3233), "Sending failed");
+    if ((!wThreadShutdown) && (isRunning)) {
+      static InfoLogger::AutoMuteToken logtokenfmq(LogWarningSupport_(3233), 1, 60);
+      theLog.log(logtokenfmq, "FMQ sending failed");
+    }
     return -1;
   }
 
@@ -1316,6 +1347,74 @@ int ConsumerFMQchannel::processForDataDistribution(DataSetReference& bc) {
   return 0;
 }
 
+int ConsumerFMQchannel::start() {
+  nTFdiscardedEOR = 0;
+
+  repackSizeStats.reset();
+  nPagesUsedForRepack = 0;
+  nPagesUsedInput = 0;
+  nIncompleteHBF = 0;
+  TFdropped = 0;
+
+  return Consumer::start();
+}
+int ConsumerFMQchannel::stop() {
+  nTFdiscardedEOR = 0;
+  isRunning = 0;
+  double timeout = 1.0; // 1s should be enough, it was tested that FMQ usually release pages every 0.5s
+
+  theLog.log(LogInfoDevel_(3003), "Consumer %s - cleaning up pending data, timeout = %.2fs", name.c_str(), timeout);
+
+  AliceO2::Common::Timer stopTimer;
+  stopTimer.reset(timeout * 1000000); // in microseconds
+
+  // wait for threads for a minimum time
+  while (!stopTimer.isTimeout()) {
+    usleep(wThreadSleepTime); // first leave a chance to update isRunning flag
+    int nRunning = 0;
+    for (auto& w : wThreads) {
+      if (w.thread != nullptr) {
+        nRunning += w.isRunning;
+      }
+    }
+    if (!nRunning) {
+      break;
+    }
+  }
+  if (senderThread) {
+    senderThreadIsRunning=1; // ensure we do another iteration now that working threads cleaned
+    while (!stopTimer.isTimeout()) {
+      usleep(wThreadSleepTime); // first leave a chance to update isRunning flag
+      if (!senderThreadIsRunning) {
+        break;
+      }
+    }
+  }
+  // FMQ release is asynchronous... wait until all pages released
+  if (gReadoutStats.counters.pagesPendingFairMQ.load() != 0) {
+    theLog.log(LogInfoDevel_(3003), "Consumer %s - waiting FMQ to release %" PRIu64 " pages", name.c_str(), gReadoutStats.counters.pagesPendingFairMQ.load());
+  }
+  while (!stopTimer.isTimeout()) {
+    if (gReadoutStats.counters.pagesPendingFairMQ.load() == 0) break;
+    usleep(wThreadSleepTime);
+  }
+
+  // report
+  theLog.log(LogInfoDevel_(3003), "Consumer %s - discarded %" PRIu64 " TFs from buffer at End Of Run", name.c_str(), nTFdiscardedEOR.load());
+
+  // log memory pool statistics
+  if (mp!=nullptr) {
+    theLog.log(LogInfoDevel_(3003), "Consumer %s - memory pool statistics ... %s", name.c_str(), mp->getStats().c_str());
+    theLog.log(LogInfoDevel_(3003), "Consumer %s - STFB repacking statistics ... number: %" PRIu64 " average page size: %" PRIu64 " max page size: %" PRIu64 " repacked/received = %" PRIu64 "/%" PRIu64 " = %.1f%%", name.c_str(), repackSizeStats.getCount(), (uint64_t)repackSizeStats.getAverage(), repackSizeStats.getMaximum(), nPagesUsedForRepack, nPagesUsedInput, nPagesUsedForRepack * 100.0 / nPagesUsedInput);
+  }
+
+  if (TFdropped) {
+    theLog.log(LogInfoSupport_(3235), "Consumer %s - %llu incomplete TF dropped", name.c_str(), (unsigned long long)TFdropped);
+  }
+
+  // wait threads completed
+  return Consumer::stop();
+}
 
 std::unique_ptr<Consumer> getUniqueConsumerFMQchannel(ConfigFile& cfg, std::string cfgEntryPoint) { return std::make_unique<ConsumerFMQchannel>(cfg, cfgEntryPoint); }
 

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.18.0"
+#define READOUT_VERSION "2.18.1"