Merge pull request #261 from sy-c/master

v2.21.1

Author: sy-c

doc/releaseNotes.md

@@ -577,3 +577,8 @@ This file describes the main feature changes for each readout.exe released versi
   - 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.
+
+## v2.21.1 - 29/08/2023
+- o2-readout-rawreader: added options logOff and dumpOrbitStats.
+- Bookkeeping: final stats publish at end of run is synchronous.
+- Monitoring: added buffers statistics in bytes: readout.bufferUsage.bytes (in addition to existing readout.bufferUsage.value, as a percentage)

src/ConsumerFMQchannel.cxx

@@ -25,7 +25,7 @@
 
 #ifdef WITH_FAIRMQ
 
-#include <fairmq/FairMQDevice.h>
+#include <fairmq/Device.h>
 #include <fairmq/FairMQMessage.h>
 #include <fairmq/FairMQTransportFactory.h>
 #include <fairmq/tools/Unique.h>
@@ -415,6 +415,7 @@ class ConsumerFMQchannel : public Consumer
       mp -> setWarningCallback(std::bind(&ConsumerFMQchannel::mplog, this, std::placeholders::_1));
       if ((mp->getId() >= 0) && (mp->getId() < ReadoutStatsMaxItems)) {
 	mp -> setBufferStateVariable(&gReadoutStats.counters.bufferUsage[mp->getId()]);
+        gReadoutStats.counters.bufferSize[mp->getId()] = memoryPoolPageSize * memoryPoolNumberOfPages;
       }
     }
     theLog.log(LogInfoDevel_(3008), "Using memory pool [%d]: %d pages x %d bytes", mp->getId(), memoryPoolNumberOfPages, memoryPoolPageSize);

src/ConsumerStats.cxx

@@ -176,8 +176,9 @@ class ConsumerStats : public Consumer
       // buffer stats
       for (int i = 0; i < ReadoutStatsMaxItems; i++) {
 	double r = snapshot.bufferUsage[i].load();
+        uint64_t b = (uint64_t)(r * snapshot.bufferSize[i].load());
 	if (r >= 0) {
-	  sendMetricNoException(Metric{(int)(r*100), "readout.bufferUsage"}.addTag(tags::Key::ID, i));
+	  sendMetricNoException(Metric{"readout.bufferUsage"}.addValue((int)(r*100), "value").addValue(b, "bytes").addTag(tags::Key::ID, i));
 	}
       }
     }

src/ReadoutEquipment.cxx

@@ -227,6 +227,7 @@ ReadoutEquipment::ReadoutEquipment(ConfigFile& cfg, std::string cfgEntryPoint, b
     mp -> setWarningCallback(std::bind(&ReadoutEquipment::mplog, this, std::placeholders::_1));
     if ((mp->getId() >= 0) && (mp->getId() < ReadoutStatsMaxItems)) {
       mp -> setBufferStateVariable(&gReadoutStats.counters.bufferUsage[mp->getId()]);
+      gReadoutStats.counters.bufferSize[mp->getId()] = memoryPoolPageSize * memoryPoolNumberOfPages;
     }
     theLog.log(LogInfoDevel_(3008), "Using memory pool [%d]: %d pages x %d bytes", mp->getId(), memoryPoolNumberOfPages, memoryPoolPageSize);
 

src/ReadoutStats.cxx

@@ -73,8 +73,11 @@ void ReadoutStats::reset(bool lightReset)
   if (!lightReset) {
     for (unsigned int i = 0; i < ReadoutStatsMaxItems; i++) {
       counters.bufferUsage[i] = -1.0;
+      counters.bufferSize[i] = 0;
     }
   }
+
+  counters.runNumber = undefinedRunNumber;
 }
 
 void ReadoutStats::print()

src/ReadoutStats.h

@@ -35,7 +35,7 @@ struct ReadoutStatsCounters {
   std::atomic<uint64_t> bytesReadout;
   std::atomic<uint64_t> bytesRecorded;
   std::atomic<uint64_t> bytesFairMQ;
-   std::atomic<double> timestamp;
+  std::atomic<double> timestamp;
   std::atomic<double> bytesReadoutRate;
   std::atomic<uint64_t> state;
   std::atomic<uint64_t> pagesPendingFairMQ;         // number of pages pending in ConsumerFMQ
@@ -48,14 +48,16 @@ struct ReadoutStatsCounters {
   std::atomic<uint32_t> logMessagesError;           // number of log messages (severity: error)
   std::atomic<uint32_t> currentOrbit;               // 1st orbit of current timeframe (last out of aggregator)
   std::atomic<double> bufferUsage[ReadoutStatsMaxItems]; // buffer usage. -1 means not used.
+  std::atomic<uint64_t> bufferSize[ReadoutStatsMaxItems]; // buffer size in bytes. 0 means not used.
   std::atomic<uint64_t> ddHBFRepacked;              // Data Distribution: number of HBF re-packed (HBF overlapping superpages)
   std::atomic<uint64_t> ddBytesCopied;              // Data Distribution: number of bytes copied (HBF overlapping superpages)
   std::atomic<uint64_t> ddMemoryPendingBytes;       // Data Distribution: number of bytes pending release in ConsumerFMQ (real memory)
   std::atomic<uint64_t> ddPayloadPendingBytes;      // Data Distribution: number of bytes pending release in ConsumerFMQ (payload only, not accounting for memory fragmentation overhead)
+  std::atomic<uint64_t> runNumber;                  // current run number (valid only in running state)
 };
 
 // version number of this struct
-const uint32_t ReadoutStatsCountersVersion = 0xA0000003;
+const uint32_t ReadoutStatsCountersVersion = 0xA0000004;
 
 // need to be able to easily transmit this struct as a whole
 static_assert(std::is_pod<ReadoutStatsCounters>::value);

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.21.0"
+#define READOUT_VERSION "2.21.1"
 

src/mainReadout.cxx

@@ -212,38 +212,71 @@ class LogbookThread {
        th = nullptr;
      }
    };
-   int publishStats() {
+   // when timeout is set, ensures the function returns only when bookkeeping operation completed
+   // otherwise, it is done asynchronously later
+   // returns 0 on success, or an error code
+   int publishStats(int timeoutMilliseconds = 0) {
      if (logbookHandle == nullptr) return __LINE__; // fail if no connection
+     AliceO2::Common::Timer timer;
+     if (timeoutMilliseconds > 0) {
+       // wait pending request completed, if any (so that we are sure to push the latest counters)
+       timer.reset(timeoutMilliseconds * 1000);
+       while (publishRequest.load()) {
+         if (timer.isTimeout()) return __LINE__;
+         usleep(1000);
+       }
+     }
      if (publishRequest.load()) return __LINE__; // fail if request already pending
      publishRequest = 1;
+     if (this->verbose) {
+       theLog.log(LogInfoDevel_(3210), "Requested to publish logbook stats");
+     }
+     if (timeoutMilliseconds > 0) {
+       // wait request completed and check status
+       while (publishRequest.load()) {
+         if (timer.isTimeout()) return __LINE__;
+         usleep(1000);
+       }
+       if (!publishSuccess.load()) {
+         return __LINE__;
+       }
+     }
      return 0;
    };
+   bool verbose = 0; // flag for extra logs on request/publish
   private:
     std::unique_ptr<o2::bkp::api::BkpClient> logbookHandle; // handle to logbook
     std::unique_ptr<std::thread> th; // a thread reading from fd and injecting to theLog
     std::atomic<int> shutdownRequest; // flag to terminate thread
     std::atomic<int> publishRequest; // flag to ask thread to publish current values
+    std::atomic<int> publishSuccess; // flag to report status of latest publish operation
     void run() {
       setThreadName("logbook");
       // thread loop, 10Hz
       while (!shutdownRequest && (logbookHandle != nullptr)) {
         if (publishRequest.load() == 1) {
+          publishSuccess = 0;
+          // copy current counters
+          ReadoutStatsCounters snapshot;
+          memcpy((void *)&snapshot, (void *)&gReadoutStats.counters, sizeof(snapshot));
           // publishing to logbook makes sense only if a run number defined
-          if (occRunNumber != undefinedRunNumber) {
-            bool isOk = false;
+          if (snapshot.runNumber.load() != undefinedRunNumber) {
             try {
               // interface: https://github.com/AliceO2Group/Bookkeeping/tree/main/cxx-client/include/BookkeepingApi
               logbookHandle->flp()->updateReadoutCountersByFlpNameAndRunNumber(
-                occRole, occRunNumber,
-                (int64_t)gReadoutStats.counters.numberOfSubtimeframes, (int64_t)gReadoutStats.counters.bytesReadout, (int64_t)gReadoutStats.counters.bytesRecorded, (int64_t)gReadoutStats.counters.bytesFairMQ
+                snapshot.source, snapshot.runNumber.load(),
+                (int64_t)snapshot.numberOfSubtimeframes.load(), (int64_t)snapshot.bytesReadout.load(), (int64_t)snapshot.bytesRecorded.load(), (int64_t)snapshot.bytesFairMQ.load()
               );
-              isOk = true;
+              if (this->verbose) {
+                theLog.log(LogInfoDevel_(3210), "Publishing logbook stats: tf = %llu, bytesReadout = %llu", (unsigned long long)snapshot.numberOfSubtimeframes.load(), (unsigned long long)snapshot.bytesReadout.load());
+              }
+              publishSuccess = 1;
             } catch (const std::exception& ex) {
               theLog.log(LogErrorDevel_(3210), "Failed to update logbook: %s", ex.what());
             } catch (...) {
               theLog.log(LogErrorDevel_(3210), "Failed to update logbook: unknown exception");
             }
-            if (!isOk) {
+            if (!publishSuccess.load()) {
               // closing logbook immediately
               logbookHandle = nullptr;
               theLog.log(LogErrorSupport_(3210), "Logbook now disabled");
@@ -405,7 +438,7 @@ class Readout
   std::unique_ptr<ReadoutDatabase> dbHandle; // handle to readout database
 #endif
 
-  void publishLogbookStats();          // publish current readout counters to logbook
+  void publishLogbookStats(int timeout = 0);          // publish current readout counters to logbook. Optional timeout in milliseconds, async op if not set.
   AliceO2::Common::Timer logbookTimer; // timer to handle readout logbook publish interval
 
   uint64_t currentTimeframeId = undefinedTimeframeId;
@@ -418,16 +451,10 @@ class Readout
 
 bool testLogbook = false; // flag for logbook test mode
 
-void Readout::publishLogbookStats()
+void Readout::publishLogbookStats(int timeout)
 {
   //  gReadoutStats.print();
 
-#ifdef WITH_LOGBOOK
-  if (theLogbookThread != nullptr) {
-    theLogbookThread->publishStats();
-  }
-#endif
-
 #ifdef WITH_DB
   if (dbHandle != nullptr) {
     dbHandle->updateRunCounters(
@@ -438,6 +465,15 @@ void Readout::publishLogbookStats()
     );
   }
 #endif
+
+#ifdef WITH_LOGBOOK
+  if (theLogbookThread != nullptr) {
+    int err = theLogbookThread->publishStats(timeout);
+    if ((timeout > 0) && (err)) {
+      theLog.log(LogErrorDevel_(3210), "Logbook publish failed within given time (%d ms)", timeout);
+    }
+  }
+#endif
 }
 
 int Readout::_init(int argc, char* argv[])
@@ -1384,6 +1420,7 @@ int Readout::_start()
   theLog.log(LogInfoSupport_(3005), "Readout executing START");
   gReadoutStats.reset(1);
   gReadoutStats.counters.state = stringToUint64("> start");
+  gReadoutStats.counters.runNumber = occRunNumber;
   gReadoutStats.counters.notify++;
   gReadoutStats.publishNow();
 
@@ -1713,8 +1750,8 @@ int Readout::_stop()
   // report log statistics
   theLog.log("Errors: %lu Warnings: %lu", theLog.getMessageCount(InfoLogger::Severity::Error), theLog.getMessageCount(InfoLogger::Severity::Warning));
 
-  // publish final logbook statistics
-  publishLogbookStats();
+  // publish final logbook statistics (synchronously with timeout)
+  publishLogbookStats(3000);
 
   // publish some final counters
   theLog.log(LogInfoDevel_(3003), "Final counters: timeframes = %" PRIu64 " readout = %s recorded = %s",

src/readRaw.cxx

@@ -21,7 +21,7 @@
 #include "CounterStats.h"
 
 //#define ERRLOG(args...) fprintf(stderr,args)
-#define ERRLOG(args...) fprintf(stdout, args)
+#define ERRLOG(args...) if (!logOff) {fprintf(stdout, args);}
 
 int main(int argc, const char* argv[])
 {
@@ -52,6 +52,11 @@ int main(int argc, const char* argv[])
   uint32_t firstTimeframeHbOrbitBegin = 0;
   bool isDefinedFirstTimeframeHbOrbitBegin = 0;
   uint32_t maxOrbit = 0;
+  uint32_t minOrbit = 0;
+  uint32_t firstOrbit = 0;
+
+  bool logOff = 0;
+  bool dumpOrbitStats =0;
 
   // parse input arguments
   // format is a list of key=value pairs
@@ -72,6 +77,8 @@ int main(int argc, const char* argv[])
       "    dumpStats=(int) : if set, some statistics are printed on HBF/TF size.\n"
       "    fileReadVerbose=(int) : if set, more information is printed when reading/decoding file.\n"
       "    timeframePeriodOrbits=(int) : if set, TF id computed (and printed, when dump enabled) for each RDH. Typically, 128 or 256.\n"
+      "    logOff=(int) : if set, logs disabled.\n"
+      "    dumpOrbitStats=(int) : if set, first / min / max orbits are printed after file read.\n"
       "    \n",
       argv[0]);
     return -1;
@@ -124,6 +131,10 @@ int main(int argc, const char* argv[])
       timeframePeriodOrbits = (uint32_t) std::stoi(value);
     } else if (key == "dumpStats") {
       dumpStats = std::stoi(value);
+    } else if (key == "logOff") {
+      logOff = std::stoi(value);
+    } else if (key == "dumpOrbitStats") {
+      dumpOrbitStats = std::stoi(value);
     } else {
       ERRLOG("unknown option %s\n", key.c_str());
     }
@@ -368,6 +379,12 @@ int main(int argc, const char* argv[])
 	  h.computeTimeframeId(firstTimeframeHbOrbitBegin, timeframePeriodOrbits);
         }
 
+        if (RDHBlockCount == 1) {
+          firstOrbit = h.getTriggerOrbit();
+          minOrbit = firstOrbit;
+          maxOrbit = firstOrbit;
+        }
+
         if (dumpRDH) {
           h.dumpRdh(pageOffset + blockOffset, 1);
         }
@@ -423,6 +440,9 @@ int main(int argc, const char* argv[])
 	if (h.getTriggerOrbit() > maxOrbit) {
 	  maxOrbit = h.getTriggerOrbit();
 	}
+	if ((h.getTriggerOrbit() < minOrbit) || (minOrbit == 0)) {
+	  minOrbit = h.getTriggerOrbit();
+	}
 
 	unsigned int linkId = h.getLinkId();
 	if (linkId <= RdhMaxLinkId) {
@@ -496,6 +516,10 @@ int main(int argc, const char* argv[])
     ERRLOG("max orbit 0x%X\n", maxOrbit);
   }
 
+  if (dumpOrbitStats) {
+    printf("Orbits:\t first = 0x%X\tmin = 0x%X\tmax = 0x%X\n", firstOrbit, minOrbit, maxOrbit);
+  }
+
   if (dumpStats) {
     // register final HBF size
     int nLinksActive = 0;