Merge pull request #188 from sy-c/master

v2.3.0

Author: sy-c

doc/releaseNotes.md

@@ -340,3 +340,8 @@ This file describes the main feature changes for each readout.exe released versi
 
 ## v2.2.0 - 07/05/2021
 - Added rate limit for eventDump: see consumer-zmq.maxRate and consumer-zmq.pagesPerBurst.
+
+## v2.3.0 - 11/05/2021
+- consumer-stats: publish/print the current timeframe Id sent to STFB.
+- consumer-stats: ZMQ stats client cleanup timeout, to avoid blocking on exit.
+- auto-mute RDH warnings: verbosity reduced if many successive logs done in a short time.

src/ConsumerFMQchannel.cxx

@@ -460,7 +460,8 @@ class ConsumerFMQchannel : public Consumer
           // printf("offset %d - HBid=%d\n",offset,lastHBid);
         }
         if (stfHeader->linkId != rdh->linkId) {
-          theLog.log(LogWarningSupport_(3004), "TF%d link Id mismatch %d != %d @ page offset %d", (int)stfHeader->timeframeId, (int)stfHeader->linkId, (int)rdh->linkId, (int)offset);
+          static InfoLogger::AutoMuteToken token(LogWarningSupport_(3004));
+          theLog.log(token, "TF%d equipment %d link Id mismatch %d != %d @ page offset %d", (int)stfHeader->timeframeId, (int)stfHeader->equipmentId, (int)stfHeader->linkId, (int)rdh->linkId, (int)offset);
           // dumpRDH(rdh);
           // printf("block %p : offset %d = %p\n",b,offset,rdh);
         }
@@ -646,6 +647,7 @@ class ConsumerFMQchannel : public Consumer
         messagesToSend.clear();
         gReadoutStats.counters.bytesFairMQ += messagesToSendSize;
         messagesToSendSize = 0;
+	gReadoutStats.counters.timeframeIdFairMQ = stfHeader->timeframeId;
       } else {
         theLog.log(LogErrorSupport_(3233), "Sending failed");
       }

src/ConsumerStats.cxx

@@ -127,6 +127,7 @@ class ConsumerStats : public Consumer
     gReadoutStats.counters.pagesPendingFairMQtime = 0;
     gReadoutStats.counters.pagesPendingFairMQreleased = 0;
     unsigned long long nRfmq = snapshot.pagesPendingFairMQreleased.load();
+    int tfidfmq = (int)snapshot.timeframeIdFairMQ.load();
     double avgTfmq = 0.0;
     double rRfmq = 0.0;
     if (nRfmq) {
@@ -157,6 +158,7 @@ class ConsumerStats : public Consumer
       sendMetricNoException({ (int)nRfmq, "readout.stfbMemoryPagesLocked"});
       sendMetricNoException({ rRfmq, "readout.stfbMemoryPagesReleaseRate"});
       sendMetricNoException({ avgTfmq, "readout.stfbMemoryPagesReleaseLatency"});
+      sendMetricNoException({ tfidfmq, "readout.stfbTimeframeId"});
     }
 
 #ifdef WITH_ZMQ
@@ -169,7 +171,7 @@ class ConsumerStats : public Consumer
     if (consoleUpdate) {
       if (deltaT > 0) {
         theLog.log(LogInfoOps_(3003), "Last interval (%.2fs): blocksRx=%llu, block rate=%.2lf, bytesRx=%llu, rate=%s", deltaT, (unsigned long long)counterBlocksDiff, counterBlocksDiff / deltaT, (unsigned long long)counterBytesDiff, NumberOfBytesToString(counterBytesDiff * 8 / deltaT, "b/s", 1000).c_str());
-        theLog.log(LogInfoOps_(3003), "STFB locked pages: current=%llu, release rate=%.2lf Hz, latency=%.3lf s", nRfmq, rRfmq, avgTfmq);
+        theLog.log(LogInfoOps_(3003), "STFB locked pages: current=%llu, release rate=%.2lf Hz, latency=%.3lf s, current TF = %d", nRfmq, rRfmq, avgTfmq, tfidfmq );
       }
     }
 
@@ -256,6 +258,12 @@ class ConsumerStats : public Consumer
           throw __LINE__;
         }
 
+	const int cfgZmqLinger = 1000;
+        zmqError = zmq_setsockopt(zmqHandle, ZMQ_LINGER, (void*)&cfgZmqLinger, sizeof(cfgZmqLinger)); // close timeout
+        if (zmqError) {
+          throw __LINE__;
+        }
+	
         zmqError = zmq_connect(zmqHandle, cfgZmqPublishAddress.c_str());
         if (zmqError) {
           throw __LINE__;

src/ReadoutEquipment.cxx

@@ -571,6 +571,8 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
     int rdhIndexInPage = 0;
     int linkId = undefinedLinkId;
 
+    static InfoLogger::AutoMuteToken logRdhErrorsToken(LogWarningSupport_(3004), 30, 5);
+
     for (size_t pageOffset = 0; pageOffset < blockSize;) {
       RdhHandle h(baseAddress + pageOffset);
       rdhIndexInPage++;
@@ -608,7 +610,7 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
       }
       if (linkId != h.getLinkId()) {
         if (cfgRdhDumpWarningEnabled) {
-          theLog.log(LogWarningSupport_(3004), "RDH #%d @ 0x%X : inconsistent link ids: %d != %d", rdhIndexInPage, (unsigned int)pageOffset, linkId, h.getLinkId());
+          theLog.log(logRdhErrorsToken, "Equipment %d RDH #%d @ 0x%X : inconsistent link ids: %d != %d", id, rdhIndexInPage, (unsigned int)pageOffset, linkId, h.getLinkId());
         }
         statsRdhCheckStreamErr++;
         break; // stop checking this page
@@ -617,7 +619,7 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
       // check no timeframe overlap in page
       if (((blockHeader.timeframeOrbitFirst < blockHeader.timeframeOrbitLast) && ((h.getTriggerOrbit() < blockHeader.timeframeOrbitFirst) || (h.getTriggerOrbit() > blockHeader.timeframeOrbitLast))) || ((blockHeader.timeframeOrbitFirst > blockHeader.timeframeOrbitLast) && ((h.getTriggerOrbit() < blockHeader.timeframeOrbitFirst) && (h.getTriggerOrbit() > blockHeader.timeframeOrbitLast)))) {
         if (cfgRdhDumpErrorEnabled) {
-          theLog.log(LogWarningSupport_(3004), "RDH #%d @ 0x%X : TimeFrame ID change in page not allowed : orbit 0x%08X not in range [0x%08X,0x%08X]", rdhIndexInPage, (unsigned int)pageOffset, (int)h.getTriggerOrbit(), (int)blockHeader.timeframeOrbitFirst, (int)blockHeader.timeframeOrbitLast);
+          theLog.log(logRdhErrorsToken, "Equipment %d RDH #%d @ 0x%X : TimeFrame ID change in page not allowed : orbit 0x%08X not in range [0x%08X,0x%08X]", id, rdhIndexInPage, (unsigned int)pageOffset, (int)h.getTriggerOrbit(), (int)blockHeader.timeframeOrbitFirst, (int)blockHeader.timeframeOrbitLast);
         }
         statsRdhCheckStreamErr++;
         break; // stop checking this page

src/ReadoutStats.cxx

@@ -41,6 +41,7 @@ void ReadoutStats::reset()
   counters.pagesPendingFairMQ = 0;
   counters.pagesPendingFairMQreleased = 0;
   counters.pagesPendingFairMQtime = 0;
+  counters.timeframeIdFairMQ = 0;
 }
 
 void ReadoutStats::print()

src/ReadoutStats.h

@@ -28,6 +28,7 @@ struct ReadoutStatsCounters {
   std::atomic<uint64_t> pagesPendingFairMQ;         // number of pages pending in ConsumerFMQ
   std::atomic<uint64_t> pagesPendingFairMQreleased; // number of pages which have been released by ConsumerFMQ
   std::atomic<uint64_t> pagesPendingFairMQtime;     // latency in FMQ, in microseconds, total for all released pages
+  std::atomic<uint32_t> timeframeIdFairMQ;          // last timeframe pushed to ConsumerFMQ
 };
 
 // need to be able to easily transmit this struct as a whole

src/ReadoutVersion.h

@@ -1 +1 @@
-#define READOUT_VERSION "2.2.0"
+#define READOUT_VERSION "2.3.0"

src/readoutMonitor.cxx

@@ -182,11 +182,11 @@ int main(int argc, const char** argv)
   double previousSampleTime = 0;
 
   // header
-  printf("               Time    State         nStf   Readout  Recorder      STFB      STFB        STFB      STFB\n");
-  printf("                                              total     total     total    memory      memory    memory\n");
-  printf("                                                                           locked     release   release\n");
-  printf("                                                                                         rate   latency\n");
-  printf("                                            (bytes)   (bytes)   (bytes)    (pages)  (pages/s)       (s)\n");
+  printf("               Time    State         nStf   Readout  Recorder      STFB      STFB        STFB      STFB      STFB\n");
+  printf("                                              total     total     total    memory      memory    memory       tf \n");
+  printf("                                                                           locked     release   release       id \n");
+  printf("                                                                                         rate   latency          \n");
+  printf("                                            (bytes)   (bytes)   (bytes)    (pages)  (pages/s)       (s)          \n");
 
   for (; !ShutdownRequest;) {
     int nb = 0;
@@ -214,7 +214,7 @@ int main(int argc, const char** argv)
 	avgTfmq = (counters->pagesPendingFairMQtime.load() / nRfmq) / (deltaT * 1000000.0);
       }
       if (cfgRawBytes) {
-        printf("%s\t%s\t%llu\t%llu\t%llu\t%llu\t%llu\t%.2lf\t%.6lf\n",
+        printf("%s\t%s\t%llu\t%llu\t%llu\t%llu\t%llu\t%.2lf\t%.6lf\t%d\n",
            t ? getStringTime(t).c_str() : "-",
            (char*)&state,
            (unsigned long long)counters->numberOfSubtimeframes.load(),
@@ -223,9 +223,10 @@ int main(int argc, const char** argv)
            (unsigned long long)counters->bytesFairMQ.load(),
            (unsigned long long)counters->pagesPendingFairMQ.load(),
            nRfmq,
-           avgTfmq);
+           avgTfmq,
+           (int)counters->timeframeIdFairMQ.load());
       } else {
-        printf("%s  %s     %8llu     %s   %s   %s   %6llu    %7.2lf    %6.4lf\n",
+        printf("%s  %s     %8llu     %s   %s   %s   %6llu    %7.2lf    %6.4lf %8d\n",
            t ? getStringTime(t).c_str() : "-",
            (char*)&state,
            (unsigned long long)counters->numberOfSubtimeframes.load(),
@@ -234,7 +235,8 @@ int main(int argc, const char** argv)
            NumberOfBytesToString(counters->bytesFairMQ.load(),"").c_str(),
            (unsigned long long)counters->pagesPendingFairMQ.load(),
            nRfmq,
-           avgTfmq);
+           avgTfmq,
+	   (int)counters->timeframeIdFairMQ.load());
       }
     }
     previousSampleTime = t;