Merge pull request #214 from sy-c/master

v2.9.1

Author: sy-c

doc/releaseNotes.md

@@ -431,3 +431,8 @@ This file describes the main feature changes for each readout.exe released versi
 
 ## v2.9.0 - 23/03/2022
 - Added optional database connection to store runtime statistics.
+
+## v2.9.1 - 30/03/2022
+- DB improvements: occ.getRole() workaround for source id, logging.
+- Added logging of some EOR stats (TF number, etc) as for bookkeeping.
+- o2-readout-monitor: indexing by source for correct rates computation.

src/ReadoutDatabase.cxx

@@ -13,7 +13,9 @@
 typedef bool my_bool;
 #endif
 
-ReadoutDatabase::ReadoutDatabase(const char* cx) {
+ReadoutDatabase::ReadoutDatabase(const char* cx, int v, const LogCallback& cb) {
+  verbose = v;
+  theLogCallback = cb;
 
   char *db_db=nullptr, *db_user=nullptr, *db_pwd=nullptr, *db_host=nullptr;
   char *p=nullptr,*ptr,*lptr;
@@ -77,9 +79,7 @@ ReadoutDatabase::ReadoutDatabase(const char* cx) {
     goto open_failed;
   }
 
-  if (verbose) {
-    printf("Using DB %s @ %s\n", db_db, db_host);
-  }
+  log("Using database " + std::string(db_db) + "@" + std::string(db_host));
 
   // try to connect
   if (mysql_real_connect(db,db_host,db_user,db_pwd,db_db,0,nullptr,0)==nullptr) {
@@ -158,9 +158,7 @@ int ReadoutDatabase::query(int maxRetry, const char *inQuery,...) {
     return -1;
   }
 
-  if (verbose) {
-    printf("Executing query: %s\n", query);
-  }
+  log("Executing query: "  + std::string(query));
 
   if (maxRetry <= 0) {
     maxRetry = 1;
@@ -298,3 +296,13 @@ const char* ReadoutDatabase::getError() {
 const char* ReadoutDatabase::getQuery() {
   return lastQuery.c_str();
 }
+
+void ReadoutDatabase::log(const std::string &msg) {
+  if (verbose) {
+    if (theLogCallback!=nullptr) {
+      theLogCallback(msg);
+    } else {
+      printf("%s\n", msg.c_str());
+    }
+  }
+}

src/ReadoutDatabase.h

@@ -3,10 +3,14 @@
 #include <stdint.h>
 #include <vector>
 #include <string>
+#include <functional>
 
 class ReadoutDatabase {
 public:
-  ReadoutDatabase(const char* cx); // create a handle to DB. Connection string in the form user:pwd@host/dbname
+  // an optional user-provided logging function for all DB-related ops
+  typedef std::function<void(const std::string &)> LogCallback;
+
+  ReadoutDatabase(const char* cx, int verbose = 0, const LogCallback& cb = nullptr); // create a handle to DB. Connection string in the form user:pwd@host/dbname
   ~ReadoutDatabase();
 
   // admin database structure
@@ -42,6 +46,9 @@ class ReadoutDatabase {
   int maxRetry = 20; // number of query retries
   int retryTimeout = 50; // retry interval (milliseconds)
 
+  LogCallback theLogCallback;
+  void log(const std::string &log);
+
   std::string lastError = ""; // error string of last query, if any
   std::string lastQuery = ""; // last query executed  
 };

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.9.0"
+#define READOUT_VERSION "2.9.1"
 

src/mainReadout.cxx

@@ -26,6 +26,7 @@
 #include <time.h>
 #include <string.h>
 #include <sys/mman.h>
+#include <inttypes.h>
 
 #include "DataBlock.h"
 #include "DataBlockContainer.h"
@@ -124,6 +125,11 @@ static void signalHandler(int signalId)
 std::string occRole;     // OCC role name
 tRunNumber occRunNumber = 0; // OCC run number
 
+// a general purpose log function for DB
+void dbLog(const std::string &msg) {
+  theLog.log(LogInfoDevel_(3012), "%s", msg.c_str());
+}
+
 class Readout
 {
 
@@ -329,22 +335,6 @@ int Readout::init(int argc, char* argv[])
     initLogs.push_back({LogWarningSupport_(3236), "Failed to start Stats publish"});
   } //otherwise: disabled
 
-  // init database
-  if (cfgDatabaseCxParams != "") {
-    #ifdef WITH_DB
-      try {
-        dbHandle=std::make_unique<ReadoutDatabase>(cfgDatabaseCxParams.c_str());
-	if (dbHandle == nullptr) { throw __LINE__; }
-	//dbHandle->verbose = 1;
-	initLogs.push_back({LogInfoSupport, "Database connected "});
-      }
-      catch(...) {
-        initLogs.push_back({LogWarningSupport_(3242), "Failed to connect database"});
-      }
-    #endif
-  }
-
-
   // configure signal handlers for clean exit
   struct sigaction signalSettings;
   bzero(&signalSettings, sizeof(signalSettings));
@@ -354,8 +344,9 @@ int Readout::init(int argc, char* argv[])
   sigaction(SIGINT, &signalSettings, NULL);
 
   // log startup and options
-  theLog.log(LogInfoSupport_(3001), "Readout " READOUT_VERSION " - process starting, pid %d", getpid());
+  theLog.log(LogInfoSupport_(3001), "Readout " READOUT_VERSION " - process starting, pid %d for role %s", getpid(), occRole.c_str());
   if (cfgVerbose) {
+    theLog.log(LogInfoDevel, "Build time: %s %s", __DATE__, __TIME__);
     theLog.log(LogInfoDevel, "Optional built features enabled:");
     #ifdef WITH_READOUTCARD
       theLog.log(LogInfoDevel, "READOUTCARD : yes");
@@ -405,12 +396,26 @@ int Readout::init(int argc, char* argv[])
       theLog.log(LogInfoDevel, "ZMQ : no");
     #endif
   }
-  
+
   // report cached logs
   for(auto const &l : initLogs) {
     theLog.log(l.first, "%s", l.second.c_str());
   }
 
+  // init database
+  if (cfgDatabaseCxParams != "") {
+    #ifdef WITH_DB
+      try {
+        dbHandle=std::make_unique<ReadoutDatabase>(cfgDatabaseCxParams.c_str(), cfgVerbose, dbLog);
+	if (dbHandle == nullptr) { throw __LINE__; }
+	theLog.log(LogInfoDevel_(3012), "Database connected ");
+      }
+      catch(...) {
+        theLog.log(LogWarningDevel_(3242), "Failed to connect database");
+      }
+    #endif
+  }
+
   return 0;
 }
 
@@ -1220,6 +1225,13 @@ int Readout::stop()
   // publish final logbook statistics
   publishLogbookStats();
 
+  // publish some final counters
+  theLog.log(LogInfoDevel_(3003), "Final counters: timeframes = %" PRIu64 " readout = %s recorded = %s",
+    gReadoutStats.counters.numberOfSubtimeframes.load(),
+    NumberOfBytesToString(gReadoutStats.counters.bytesReadout.load(), "bytes").c_str(),
+    NumberOfBytesToString(gReadoutStats.counters.bytesRecorded.load(),"bytes").c_str()
+  );
+
   theLog.log(LogInfoSupport_(3005), "Readout completed STOP");
   gReadoutStats.counters.state = stringToUint64("ready");
   gReadoutStats.counters.notify++;
@@ -1324,14 +1336,20 @@ class ReadoutOCCStateMachine : public RuntimeControlledObject
   ReadoutOCCStateMachine(std::unique_ptr<Readout> r) : RuntimeControlledObject("Readout Process")
   {
     theReadout = std::move(r);
-    occRole = this->getRole();
+    // the following does not work: getRole() is empty at this stage - BUG. O2_ROLE is defined.
+    // occRole = this->getRole();
   }
 
   int executeConfigure(const boost::property_tree::ptree& properties)
   {
     if (theReadout == nullptr) {
       return -1;
     }
+
+    if (this->getRole() != occRole) {
+      theLog.log(LogWarningDevel_(3243), "OCC role mismatch: getRole()=%s %s=%s occRole=%s", this->getRole().c_str(), OCC_ROLE_ENV, getenv(OCC_ROLE_ENV), occRole.c_str());
+    }
+
     return theReadout->configure(properties);
   }
 

src/readoutAdminDB.cxx

@@ -26,7 +26,7 @@ using namespace AliceO2::InfoLogger;
 void printUsage()
 {
   printf("Usage: readoutAdminDB ...\n");
-  printf("  -c command : action to execute. One of create (create tables), clear (delete tables content), destroy (destroy all tables), test (insert some dummy data)\n");  
+  printf("  -c command : action to execute. One of create (create tables), clear (delete tables content), destroy (destroy all tables), fetch (retrieve content), test (insert some dummy data)\n");
   printf("  [-z pathToConfigurationFile] : sets which configuration to use. By default %s\n", cfgDefaultsPath.c_str());
   printf("  [-v] : sets verbose mode\n");
   printf("  [-h] : print this help\n");

src/readoutErrorCodes.h

@@ -26,6 +26,7 @@
   { 3009, "Trace for readout data dump", nullptr},
   { 3010, "Trace for readout hardware operations and status", nullptr},
   { 3011, "Trace for readout data stream content", nullptr},
+  { 3012, "Trace for readout database operations", nullptr},
 
   { 3100, "Configuration problem", nullptr},
   { 3101, "A feature is configured but not supported by this readout build.", nullptr},
@@ -47,5 +48,6 @@
   { 3240, "Readout card problem", nullptr},
   { 3241, "Inconsistent data", nullptr},
   { 3242, "Database error", nullptr},
+  { 3243, "Control problem", nullptr},
 */
 

src/readoutMonitor.cxx

@@ -12,6 +12,7 @@
 #include <Common/Configuration.h>
 #include <InfoLogger/InfoLogger.hxx>
 #include <InfoLogger/InfoLoggerMacros.hxx>
+#include <map>
 #include <signal.h>
 #include <zmq.h>
 #include "ReadoutStats.h"
@@ -214,7 +215,9 @@ int main(int argc, const char** argv)
   sigaction(SIGINT, &signalSettings, NULL);
 
   theLog.log(LogInfoDevel_(3006), "Entering monitoring loop");
-  double previousSampleTime = 0;
+
+  typedef std::map<std::string, double> MapCountersBySource;
+  MapCountersBySource latestUpdate; // keep copy of latest counter for each source
 
   // header
   if ((!cfgRawBytes)&&(broadcastSocket == nullptr)) {
@@ -244,6 +247,20 @@ int main(int argc, const char** argv)
     double t = counters->timestamp.load();
     double nRfmq = counters->pagesPendingFairMQreleased.load();
     double avgTfmq = 0.0;
+
+    std::string k = counters->source;
+    MapCountersBySource::iterator lb = latestUpdate.lower_bound(k);
+    double previousSampleTime = 0;
+
+    if(lb != latestUpdate.end() && !(latestUpdate.key_comp()(k, lb->first))) {
+      // source was already sampled
+      previousSampleTime = lb->second;
+      lb->second = t;
+    } else {
+      // this is first sample for this source
+      latestUpdate.insert(lb, MapCountersBySource::value_type(k, t));
+    }
+
     if (previousSampleTime > 0) {
       double deltaT = t - previousSampleTime;
       nRfmq = nRfmq / deltaT;

src/readoutStatus.tcl

@@ -12,6 +12,7 @@ set configSection "readout-monitor"
 # defaults
 set cfgValue_broadcastPort ""
 set cfgValue_broadcastHost ""
+set cfgValue_verbose 0
 
 if {[llength $argv] >=2} {
   set configFile [lindex $argv 0]
@@ -484,7 +485,10 @@ proc processInput {} {
     }
 
     if {[gets $server_fd msg]==-1} {break}
-    puts "Msg: $msg"
+    global cfgValue_verbose
+    if {$cfgValue_verbose} {
+      puts "Msg: $msg"
+    }
     set lm [split $msg "\t"]
     if {[llength $lm]!=11} {break}
     updateNode $lm