Merge pull request #200 from sy-c/master

v2.6

Author: sy-c

doc/releaseNotes.md

@@ -380,3 +380,8 @@ This file describes the main feature changes for each readout.exe released versi
 ## v2.5.2 - 13/09/2021
 - Updated configuration parameters:
   - equipment.TFperiod is now set to 128 by default, instead of 256 previously. This is the duration of a timeframe, in number of LHC orbits.
+
+## v2.6 - 06/10/2021
+- Built-in memory usage reporting for o2-readout-test-fmq-memory
+- Added a default configuration file (/etc/o2.d/readout-defaults.cfg) to store some global settings, loaded on process startup. This is independent from the CONFIGURE time configuration file.
+- Process memory lock is now disabled by default (saves 2GB of unused FMQ virtual memory created for each channel).

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.5.2"
+#define READOUT_VERSION "2.6"
 

src/mainReadout.cxx

@@ -154,6 +154,7 @@ class Readout
   std::string cfgLogbookApiToken;
   int cfgLogbookUpdateInterval;
   std::string cfgTimeframeServerUrl;
+  int cfgVerbose = 0;
 
   // runtime entities
   std::vector<std::unique_ptr<Consumer>> dataConsumers;
@@ -226,6 +227,59 @@ void Readout::publishLogbookStats()
 
 int Readout::init(int argc, char* argv[])
 {
+  int doMemLock = 0; // when set, ensure all allocated memory is locked in ram
+  std::string readoutExe = ""; // when set, use specified executable
+  std::string readoutConfig = ""; // when set, use specified executable
+
+  // cache of logs - delay startup messages
+  std::vector<std::pair<AliceO2::InfoLogger::InfoLogger::InfoLoggerMessageOption, std::string>> initLogs;
+
+  // load configuration defaults
+  ConfigFile cfgDefaults;
+  const std::string cfgDefaultsPath = "file:/etc/o2.d/readout-defaults.cfg"; // path to default configuration file
+  const std::string cfgDefaultsEntryPoint = "readout"; // entry point for default configuration variables (e.g. section named [readout])
+  try {
+    cfgDefaults.load(cfgDefaultsPath.c_str());
+    initLogs.push_back({LogInfoDevel, "Defaults loaded from " + cfgDefaultsPath});
+    cfgDefaults.getOptionalValue<int>(cfgDefaultsEntryPoint + ".memLock", doMemLock, doMemLock);
+    cfgDefaults.getOptionalValue<std::string>(cfgDefaultsEntryPoint + ".readoutExe", readoutExe, readoutExe);
+    cfgDefaults.getOptionalValue<std::string>(cfgDefaultsEntryPoint + ".readoutConfig", readoutConfig, readoutConfig);
+    cfgDefaults.getOptionalValue<int>(cfgDefaultsEntryPoint + ".verbose", cfgVerbose, cfgVerbose);
+  }
+  catch(...) {
+    //initLogs.push_back({LogWarningSupport_(3100), std::string("Error loading defaults")});
+  }
+
+  // redirect executable (if different from self!)
+  if (readoutExe.length() && readoutExe!=argv[0]) {
+    std::vector<char*> argv2;
+    argv2.push_back((char *)readoutExe.c_str());
+    if (readoutConfig.length()) {
+      argv2.push_back((char *)readoutConfig.c_str());
+    }
+    for (int i=argv2.size(); i<argc; i++) {
+      argv2.push_back(argv[i]);
+    }
+    printf("Launching ");
+    for (auto const &a : argv2) {
+      printf("%s ",a);
+    }
+    printf("\n");
+    argv2.push_back(NULL);
+    execv(readoutExe.c_str(), &argv2[0]);
+    printf("Failed to execute : %s\n",strerror(errno));
+    exit(1);
+  }
+
+  // before anything, ensure all memory used by readout is kept in RAM
+  if (doMemLock) {
+    if (mlockall(MCL_CURRENT | MCL_FUTURE) == 0) {
+      initLogs.push_back({LogInfoDevel, "Memory locked"});
+    } else {
+      initLogs.push_back({LogWarningSupport_(3230), "Failed to lock memory"});
+    }
+  }
+
   if (argc < 2) {
     printf("Please provide path to configuration file\n");
     return -1;
@@ -245,49 +299,56 @@ int Readout::init(int argc, char* argv[])
 
   // log startup and options
   theLog.log(LogInfoSupport_(3001), "Readout " READOUT_VERSION " - process starting, pid %d", getpid());
-  theLog.log(LogInfoDevel, "Optional built features enabled:");
-#ifdef WITH_READOUTCARD
-  theLog.log(LogInfoDevel, "READOUTCARD : yes");
-#else
-  theLog.log(LogInfoDevel, "READOUTCARD : no");
-#endif
-#ifdef WITH_CONFIG
-  theLog.log(LogInfoDevel, "CONFIG : yes");
-#else
-  theLog.log(LogInfoDevel, "CONFIG : no");
-#endif
-#ifdef WITH_FAIRMQ
-  theLog.log(LogInfoDevel, "FAIRMQ : yes");
-  // redirect FMQ logs to infologger
-  setFMQLogsToInfoLogger(&theLog);
-#else
-  theLog.log(LogInfoDevel, "FAIRMQ : no");
-#endif
-#ifdef WITH_NUMA
-  theLog.log(LogInfoDevel, "NUMA : yes");
-#else
-  theLog.log(LogInfoDevel, "NUMA : no");
-#endif
-#ifdef WITH_RDMA
-  theLog.log(LogInfoDevel, "RDMA : yes");
-#else
-  theLog.log(LogInfoDevel, "RDMA : no");
-#endif
-#ifdef WITH_OCC
-  theLog.log(LogInfoDevel, "OCC : yes");
-#else
-  theLog.log(LogInfoDevel, "OCC : no");
-#endif
-#ifdef WITH_LOGBOOK
-  theLog.log(LogInfoDevel, "LOGBOOK : yes");
-#else
-  theLog.log(LogInfoDevel, "LOGBOOK : no");
-#endif
-#ifdef WITH_ZMQ
-  theLog.log(LogInfoDevel, "ZMQ : yes");
-#else
-  theLog.log(LogInfoDevel, "ZMQ : no");
-#endif
+  if (cfgVerbose) {
+    theLog.log(LogInfoDevel, "Optional built features enabled:");
+    #ifdef WITH_READOUTCARD
+      theLog.log(LogInfoDevel, "READOUTCARD : yes");
+    #else
+      theLog.log(LogInfoDevel, "READOUTCARD : no");
+    #endif
+    #ifdef WITH_CONFIG
+      theLog.log(LogInfoDevel, "CONFIG : yes");
+    #else
+      theLog.log(LogInfoDevel, "CONFIG : no");
+    #endif
+    #ifdef WITH_FAIRMQ
+      theLog.log(LogInfoDevel, "FAIRMQ : yes");
+      // redirect FMQ logs to infologger
+      setFMQLogsToInfoLogger(&theLog);
+    #else
+      theLog.log(LogInfoDevel, "FAIRMQ : no");
+    #endif
+    #ifdef WITH_NUMA
+      theLog.log(LogInfoDevel, "NUMA : yes");
+    #else
+      theLog.log(LogInfoDevel, "NUMA : no");
+    #endif
+    #ifdef WITH_RDMA
+      theLog.log(LogInfoDevel, "RDMA : yes");
+    #else
+      theLog.log(LogInfoDevel, "RDMA : no");
+    #endif
+    #ifdef WITH_OCC
+      theLog.log(LogInfoDevel, "OCC : yes");
+    #else
+      theLog.log(LogInfoDevel, "OCC : no");
+    #endif
+    #ifdef WITH_LOGBOOK
+      theLog.log(LogInfoDevel, "LOGBOOK : yes");
+    #else
+      theLog.log(LogInfoDevel, "LOGBOOK : no");
+    #endif
+    #ifdef WITH_ZMQ
+      theLog.log(LogInfoDevel, "ZMQ : yes");
+    #else
+      theLog.log(LogInfoDevel, "ZMQ : no");
+    #endif
+  }
+  
+  // report cached logs
+  for(auto const &l : initLogs) {
+    theLog.log(l.first, "%s", l.second.c_str());
+  }
 
   return 0;
 }
@@ -1250,12 +1311,8 @@ class ReadoutOCCStateMachine : public RuntimeControlledObject
 // the main program loop
 int main(int argc, char* argv[])
 {
-  // before anything, ensure all memory used by readout is kept in RAM
-  bool memLockSuccess = false;
-  if (mlockall(MCL_CURRENT | MCL_FUTURE) == 0) {
-    memLockSuccess = true;
-  }
-
+  //printf("Starting %s - %d\n",argv[0], (int)getpid());
+  
   // check environment settings
 
   // OCC control port. If set, use OCClib to handle Readout states.
@@ -1285,11 +1342,6 @@ int main(int argc, char* argv[])
     return err;
   }
 
-  // report memlock status
-  if (!memLockSuccess) {
-    theLog.log(LogWarningSupport_(3230), "Failed to lock readout memory");
-  }
-
   if (occMode) {
 #ifdef WITH_OCC
     theLog.log(LogInfoDevel, "Readout entering OCC state machine");

src/testFmqMemory.cxx

@@ -14,14 +14,41 @@
 #include <string.h>
 #include <chrono>
 #include <sys/types.h>
+#include <sys/mman.h>
+#include <unistd.h>
 
 #include <fairmq/FairMQDevice.h>
 #include <fairmq/FairMQTransportFactory.h>
 //#include <fairmq/tools/Unique.h>
 
+void logMemoryUsage() {
+  double memPageSize = getpagesize() / (1024.0*1024.0);
+  const int maxpath = 256;
+  char fn[maxpath];
+  snprintf(fn, maxpath, "/proc/%d/statm", getpid());
+  FILE *fp=fopen(fn,"r");
+  const int maxline = 256;
+  char buf[maxline];
+  if (fgets(buf, maxline, fp) != NULL) {
+    int vsize, vresident, vshared, vtext, vlib, vdata, vdt;
+    if (sscanf(buf, "%d %d %d %d %d %d %d", &vsize, &vresident, &vshared, &vtext, &vlib, &vdata, &vdt) == 7) {    
+      printf("Memory stats: size = %6.2f MB\tresident = %6.2f MB\tshared = %6.2f MB\n", vsize * memPageSize, vresident * memPageSize, vshared * memPageSize);
+    }
+  }
+  fclose(fp);
+}
+
+
 #define GB *1073741824LLU
 #define SLEEPTIME 3
-#define WAITHERE printf("Waiting %ds ",SLEEPTIME); for(int k=0; k<SLEEPTIME; k++) {printf(".");fflush(stdout);usleep(1000000);} printf("\n");
+//#define WAITHERE printf("Waiting %ds ",SLEEPTIME); for(int k=0; k<SLEEPTIME; k++) {printf(".");fflush(stdout);usleep(1000000);} printf("\n");
+#define WAITHERE logMemoryUsage();
+
+
+// memory settings
+const bool memlock = false;	// lock the whole process memory
+const bool fmqMemLock = true;	// lock FMQ region
+const bool fmqMemZero = false;	// zero FMQ region
 
 
 int main(int argc, const char* argv[]) {
@@ -34,7 +61,14 @@ int main(int argc, const char* argv[]) {
     return -1;
   }
 
-  printf("Startup %d\n",(int)getpid());
+  printf("Locking process memory: %s\n", memlock ? "yes" : "no");
+  if (memlock) {
+    if (mlockall(MCL_CURRENT | MCL_FUTURE) != 0) {
+      printf("failed to lock memory\n");
+    }
+  }
+
+  printf("Startup pid %d\n",(int)getpid());
   WAITHERE;
 
   std::unique_ptr<FairMQChannel> sendingChannel;
@@ -48,13 +82,13 @@ int main(int argc, const char* argv[]) {
   sendingChannel = std::make_unique<FairMQChannel>("readout-test", "pair", transportFactory);
   WAITHERE;
 
-  printf("Get unmanaged memory\n");
+  printf("Get unmanaged memory (lock=%s, zero=%s)\n", fmqMemLock ? "yes" : "no", fmqMemZero ? "yes" : "no");
   long long mMemorySize = ngb GB;
   auto t00 = std::chrono::steady_clock::now();
   try {
 //  memoryBuffer = sendingChannel->Transport()->CreateUnmanagedRegion(mMemorySize, [](void* /*data*/, size_t /*size*/, void* /*hint*/) {});
   memoryBuffer = sendingChannel->Transport()->CreateUnmanagedRegion(mMemorySize, [](void* /*data*/, size_t /*size*/, void* /*hint*/) {},
-  "",0,fair::mq::RegionConfig{true, false}); // lock / zero
+  "",0,fair::mq::RegionConfig{fmqMemLock, fmqMemZero}); // lock / zero
   }
   catch(...) {
     printf("Failed to get buffer (exception)\n"); return 1; 
@@ -66,6 +100,7 @@ int main(int argc, const char* argv[]) {
   WAITHERE;
 
   printf("Write to memory, by chunks of 1GB\n");
+  t00 = std::chrono::steady_clock::now();
   for (unsigned int i=0; i<ngb; i++) {
     auto t0 = std::chrono::steady_clock::now();
     char *ptr=(char *)memoryBuffer->GetData();
@@ -82,11 +117,10 @@ int main(int argc, const char* argv[]) {
     std::chrono::duration<double> tdiff = std::chrono::steady_clock::now() - t0;
     printf(" %.2lf GB/s\n", 1.0/tdiff.count());
   }
+  std::chrono::duration<double> tdiff1 = std::chrono::steady_clock::now() - t00;
   printf("Done writing\n");
-
-  tdiff0 = std::chrono::steady_clock::now() - t00;
-  printf("Average: %.2lf GB/s (including alloc)\n", ngb * 1.0/(tdiff0.count()-SLEEPTIME));
-
+  printf("Average: %.2lf GB/s (writing)\n", ngb * 1.0/tdiff1.count());
+  printf("Average: %.2lf GB/s (writing + malloc)\n", ngb * 1.0/(tdiff0.count() + tdiff1.count()));
   WAITHERE;
 
   printf("Cleanup FMQ unmanaged region\n");