Support upload of the chunked Zorro objects (AliceO2Group#8001)

Author: mpuccio

EventFiltering/macros/uploadOTSobjects.C

@@ -16,6 +16,7 @@
 #include <string>
 #include <map>
 #include <array>
+#include <cmath>
 
 #include "TFile.h"
 #include "TGrid.h"
@@ -26,16 +27,23 @@
 
 #include "CCDB/BasicCCDBManager.h"
 #include "EventFiltering/ZorroHelper.h"
+#include "CommonConstants/LHCConstants.h"
 
-void uploadOTSobjects(std::string inputList, std::string passName, bool useAlien)
+constexpr uint32_t chunkSize = 1000000;
+
+void uploadOTSobjects(std::string inputList, std::string passName, bool useAlien, bool chunkedProcessing)
 {
   const std::string kBaseCCDBPath = "Users/m/mpuccio/EventFiltering/OTS/";
   std::string baseCCDBpath = passName.empty() ? kBaseCCDBPath : kBaseCCDBPath + passName + "/";
+  if (chunkedProcessing) {
+    baseCCDBpath += "Chunked/";
+  }
   if (useAlien) {
     TGrid::Connect("alien://");
   }
   o2::ccdb::CcdbApi api;
   api.init("http://alice-ccdb.cern.ch");
+  auto& ccdb = o2::ccdb::BasicCCDBManager::instance();
 
   std::ifstream file(inputList.data());
   std::string path;
@@ -47,9 +55,11 @@ void uploadOTSobjects(std::string inputList, std::string passName, bool useAlien
     const int runNumber = std::stoi(runString);
     metadata["runNumber"] = runString;
     std::pair<int64_t, int64_t> duration = o2::ccdb::BasicCCDBManager::getRunDuration(api, runNumber);
-    duration.first -= 10000;   // subtract 3 minutes from the run start
-    duration.second += 180000; // add 3 minutes to the run duration
+    duration.first -= 60000;  // subtract 1 minutes from the run start
+    duration.second += 60000; // add 1 minutes to the run duration
     std::cout << ">>> Begin - end timestamps for the upload: " << duration.first << " - " << duration.second << std::endl;
+    auto ctp = ccdb.getForTimeStamp<std::vector<Long64_t>>("CTP/Calib/OrbitReset", duration.first / 2 + duration.second / 2);
+    auto orbitResetTimestamp = (*ctp)[0];
     path = useAlien ? "alien://" + path : path;
     std::unique_ptr<TFile> scalersFile{TFile::Open((path + "/AnalysisResults_fullrun.root").data(), "READ")};
     TH1* scalers = static_cast<TH1*>(scalersFile->Get("central-event-filter-task/scalers/mScalers"));
@@ -85,13 +95,48 @@ void uploadOTSobjects(std::string inputList, std::string passName, bool useAlien
         }
       }
     }
-    api.storeAsTFileAny(&zorroHelpers, baseCCDBpath + "ZorroHelpers", metadata, duration.first, duration.second);
+    std::sort(zorroHelpers.begin(), zorroHelpers.end(), [](const ZorroHelper& a, const ZorroHelper& b) {
+      return a.bcAOD < b.bcAOD;
+    });
+    if (!chunkedProcessing) {
+      api.storeAsTFileAny(&zorroHelpers, baseCCDBpath + "ZorroHelpers", metadata, duration.first, duration.second);
+      std::cout << std::endl;
+    } else {
+      uint32_t helperIndex{0};
+      auto startTS = duration.first;
+      int64_t endTS = 0;
+      while (helperIndex < zorroHelpers.size()) {
+        std::vector<ZorroHelper> chunk;
+        uint32_t endIndex = helperIndex + chunkSize;
+        while (true) {
+          if (endIndex >= zorroHelpers.size() - 2) {
+            endTS = duration.second;
+            chunk.insert(chunk.begin(), zorroHelpers.begin() + helperIndex, zorroHelpers.end());
+            break;
+          }
+          auto bcEnd{zorroHelpers[endIndex].bcAOD > zorroHelpers[endIndex].bcEvSel ? zorroHelpers[endIndex].bcAOD : zorroHelpers[endIndex].bcEvSel};
+          const auto& nextHelper = zorroHelpers[endIndex + 1];
+          auto bcNext{nextHelper.bcAOD < nextHelper.bcEvSel ? nextHelper.bcAOD : nextHelper.bcEvSel};
+          if (bcNext - bcEnd > 2001 / o2::constants::lhc::LHCBunchSpacingMUS) { /// ensure a gap of 2ms between chunks
+            chunk.insert(chunk.begin(), zorroHelpers.begin() + helperIndex, zorroHelpers.begin() + endIndex + 1);
+            endTS = (orbitResetTimestamp + int64_t(bcEnd * o2::constants::lhc::LHCBunchSpacingNS * 1e-3)) / 1000 + 1;
+            break;
+          }
+          bcEnd = nextHelper.bcAOD > nextHelper.bcEvSel ? nextHelper.bcAOD : nextHelper.bcEvSel;
+          endIndex++;
+        }
+        std::cout << ">>> Chunk " << helperIndex << " - " << helperIndex + chunk.size() << " : " << startTS << " - " << endTS << " \t" << (endTS - startTS) * 1.e-3 << std::endl;
+        api.storeAsTFileAny(&zorroHelpers, baseCCDBpath + "ZorroHelpers", metadata, startTS, endTS);
+        startTS = endTS + 1;
+        helperIndex += chunk.size();
+      }
+    }
   }
 }
 
-void uploadOTSobjects(std::string periodName)
+void uploadOTSobjects(std::string periodName, bool chunkedProcessing)
 {
   int year = 2000 + std::stoi(periodName.substr(3, 2));
   gSystem->Exec(Form("alien_find /alice/data/%i/%s/ ctf_skim_full/AnalysisResults_fullrun.root | sed 's:/AnalysisResults_fullrun\\.root::' > list_%s.txt", year, periodName.data(), periodName.data()));
-  uploadOTSobjects(Form("list_%s.txt", periodName.data()), "", true);
+  uploadOTSobjects(Form("list_%s.txt", periodName.data()), "", true, chunkedProcessing);
 }