stfb tfid publish

Author: sy-c

doc/releaseNotes.md

@@ -340,3 +340,6 @@ 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.
+
+## next version
+- consumer-stats: publish/print the current timeframe Id sent to STFB.

src/ConsumerFMQchannel.cxx

@@ -646,6 +646,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 );
       }
     }
 

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/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;