Use pool CDBManagers to emulate parallel proc. of many TFs

Author: shahor02

Detectors/Calibration/testMacros/populateCCDB.cxx

@@ -21,7 +21,7 @@ namespace bpo = boost::program_options;
 bool initOptionsAndParse(bpo::options_description& options, int argc, char* argv[], bpo::variables_map& vm)
 {
   options.add_options()(
-    "ccdb-server,s", bpo::value<std::string>()->default_value("ccdb-test.cern.ch:8080"), "CCDB server")(
+    "ccdb-server,s", bpo::value<std::string>()->default_value("http://ccdb-test.cern.ch:8080"), "CCDB server")(
     "in-file-name,n", bpo::value<std::string>()->default_value("cdbSizeV0.txt"), "File name with list of CCDB entries to upload")(
     "help,h", "Produce help message.");
 

Detectors/Calibration/testMacros/retrieveFromCCDB.C

@@ -11,35 +11,48 @@
 #include "CCDB/BasicCCDBManager.h"
 #include <pthread.h>
 #include <thread>
-#include <mutex>
 #endif
 #include "populateCCDB.C"
 
-using CcdbManager = o2::ccdb::BasicCCDBManager;
+using CcdbManager = o2::ccdb::CCDBManagerInstance;
+
+std::vector<std::unique_ptr<CcdbManager>> ccdbManPool;
+std::vector<bool> ccdbManPoolFlag;
+std::decay_t<decltype(std::chrono::high_resolution_clock::now())> startTime{};
 
 // macro to populate the CCDB emulating the rates that we expect for
 // Run 3, as read (in terms of size and rate) from an external file
 
-void retrieve(const std::vector<CCDBObj>& objs, float procTimeSec, std::mutex& mtx, std::atomic<int>& n, size_t tfID);
+int getFreeCCDBManagerID();
+void retrieve(const std::vector<CCDBObj>& objs, float procTimeSec, int ccdbID, std::atomic<int>& n, size_t tfID);
 
 void retrieveFromCCDB(int maxTFs = 8, float procTimeSec = 10.,
                       const std::string& fname = "cdbSizeV0.txt",
                       const std::string& ccdbHost = "http://localhost:8080" /*"http://ccdb-test.cern.ch:8080"*/,
                       bool allowCaching = true)
 {
-  auto& mgr = CcdbManager::instance();
-  mgr.setURL(ccdbHost.c_str()); // or http://localhost:8080 for a local installation
-  mgr.setCaching(allowCaching);
+  ccdbManPool.resize(1 + maxTFs);
+  ccdbManPoolFlag.resize(ccdbManPool.size(), false); // nothig is used at the moment
+  LOG(INFO) << "Caching is " << (allowCaching ? "ON" : "OFF");
+  for (auto& mgr : ccdbManPool) {
+    mgr = std::make_unique<CcdbManager>(ccdbHost.c_str());
+    mgr->setCaching(allowCaching);
+  }
   auto objs = readObjectsList(fname);
   if (objs.empty()) {
     return;
   }
-  std::mutex ccdb_mtx;
+  startTime = std::chrono::high_resolution_clock::now();
+
   std::atomic<int> nTFs{0};
   size_t tfID = 0;
   while (1) {
     if (nTFs < maxTFs) {
-      std::thread th(retrieve, std::cref(objs), procTimeSec, std::ref(ccdb_mtx), std::ref(nTFs), tfID++);
+      int ccdbID = getFreeCCDBManagerID();
+      if (ccdbID < 0) { // all slots are busy, wait
+        continue;
+      }
+      std::thread th(retrieve, std::cref(objs), procTimeSec, ccdbID, std::ref(nTFs), tfID++);
       th.detach();
       LOG(INFO) << nTFs << " TFs currently in processing";
     } else {
@@ -48,28 +61,43 @@ void retrieveFromCCDB(int maxTFs = 8, float procTimeSec = 10.,
   }
 }
 
-void retrieve(const std::vector<CCDBObj>& objs, float procTimeSec, std::mutex& mtx, std::atomic<int>& n, size_t tfID)
+void retrieve(const std::vector<CCDBObj>& objs, float procTimeSec, int ccdbID, std::atomic<int>& n, size_t tfID)
 {
   // function to retrieve the CCDB objects and wait some (TF processing) time
   // to avoid all treads starting to read at the same time, we randomize the time of reading within the allocated processing time
-
+  auto mgr = ccdbManPool[ccdbID].get();
   n++;
-  auto& mgr = CcdbManager::instance();
   float tfrac = gRandom->Rndm();
   usleep(long(procTimeSec * tfrac * 1e6));
+  auto ret_start = std::chrono::high_resolution_clock::now();
   for (auto& o : objs) {
-    auto now = std::chrono::system_clock::now();
+    auto now = std::chrono::high_resolution_clock::now();
     auto now_ms = std::chrono::time_point_cast<std::chrono::milliseconds>(now);
     auto timeStamp = now_ms.time_since_epoch();
-    mtx.lock();
-    std::vector<uint8_t>* ob = mgr.getForTimeStamp<std::vector<uint8_t>>(o.path, timeStamp.count());
-    mtx.unlock();
-    LOG(INFO) << "Retrieved object " << o.path << " of size " << ob->size() << " Bytes"
-              << " for TF " << tfID;
-    if (!mgr.isCachingEnabled()) { // we can delete object only when caching is disabled
+    std::vector<uint8_t>* ob = mgr->getForTimeStamp<std::vector<uint8_t>>(o.path, timeStamp.count());
+    LOG(DEBUG) << "Retrieved object " << o.path << " of size " << ob->size() << " Bytes"
+               << " for TF " << tfID;
+    if (!mgr->isCachingEnabled()) { // we can delete object only when caching is disabled
       delete ob;
     }
   }
+  auto ret_end = std::chrono::high_resolution_clock::now();
   usleep(long(procTimeSec * (1. - tfrac) * 1e6));
+  std::chrono::duration<double, std::ratio<1, 1>> elapsedSeconds = std::chrono::high_resolution_clock::now() - startTime;
+  std::chrono::duration<double, std::ratio<1, 1>> retTime = ret_end - ret_start;
+  LOG(INFO) << "Finished TF " << tfID << " elapsed time " << elapsedSeconds.count() << " s., CCDB query time: " << retTime.count() << " s.";
   n--;
+  ccdbManPoolFlag[ccdbID] = false; // release the manager
+}
+
+int getFreeCCDBManagerID()
+{
+  /// get ID of 1st CCDBManager not used by any thread
+  for (unsigned i = 0; i < ccdbManPoolFlag.size(); i++) {
+    if (!ccdbManPoolFlag[i]) {
+      ccdbManPoolFlag[i] = true;
+      return i;
+    }
+  }
+  return -1;
 }

Detectors/Calibration/testMacros/retrieveFromCCDB.cxx

@@ -23,7 +23,7 @@ bool initOptionsAndParse(bpo::options_description& options, int argc, char* argv
   options.add_options()(
     "TFs-in-parallel,m", bpo::value<int>()->default_value(8), "Number of TFs to simulate that access the CCDB in parallel")(
     "TF-processing-time,t", bpo::value<float>()->default_value(10.), "Seconds supposed to be needed to process a TF")(
-    "ccdb-sercer,s", bpo::value<std::string>()->default_value("ccdb-test.cern.ch:8080"), "CCDB server")(
+    "ccdb-sercer,s", bpo::value<std::string>()->default_value("http://ccdb-test.cern.ch:8080"), "CCDB server")(
     "in-file-name,n", bpo::value<std::string>()->default_value("cdbSizeV0.txt"), "File name with list of CCDB entries to upload")(
     "disable-caching,d", bpo::value<bool>()->default_value(false)->implicit_value(true), "Disable CCDB caching")(
     "help,h", "Produce help message.");
@@ -62,7 +62,7 @@ int main(int argc, char* argv[])
   auto& inputFile = vm["in-file-name"].as<std::string>();
   auto& ccdbHost = vm["ccdb-sercer"].as<std::string>();
   auto disableCaching = vm["disable-caching"].as<bool>();
-  retrieveFromCCDB(nTFs, tTF, inputFile, ccdbHost, disableCaching);
+  retrieveFromCCDB(nTFs, tTF, inputFile, ccdbHost, !disableCaching);
 
   return (0);
 }