CTF: add shuffle TF option for debug

Author: f3sch

Detectors/CTF/workflow/include/CTFWorkflow/CTFReaderSpec.h

@@ -47,6 +47,7 @@ struct CTFReaderInp {
   unsigned int decSSpecEMC = 0;
   int tfRateLimit = -999;
   size_t minSHM = 0;
+  bool shuffle{false};
 };
 
 /// create a processor spec

Detectors/CTF/workflow/src/CTFReaderSpec.cxx

@@ -11,9 +11,14 @@
 
 /// @file   CTFReaderSpec.cxx
 
+#include <random>
 #include <vector>
+#include <algorithm>
+#include <numeric>
+
 #include <TFile.h>
 #include <TTree.h>
+#include <TStopwatch.h>
 
 #include "Framework/Logger.h"
 #include "Framework/ControlService.h"
@@ -49,7 +54,6 @@
 #include "CCDB/BasicCCDBManager.h"
 #include "CommonConstants/LHCConstants.h"
 #include "Algorithm/RangeTokenizer.h"
-#include <TStopwatch.h>
 #include <fairmq/Device.h>
 
 using namespace o2::framework;
@@ -299,6 +303,25 @@ void CTFReaderSpec::openCTFFile(const std::string& flname)
     if (mCTFTree->GetEntries() < 1) {
       throw std::runtime_error(fmt::format("CTF tree in {} has 0 entries, skipping", flname));
     }
+    if (mInput.shuffle) {
+      std::random_device dev;
+      std::mt19937 gen{dev()};
+      int entries = mCTFTree->GetEntries();
+      if (mInput.maxTFs > 0) {
+        entries = std::min(entries, mInput.maxTFs);
+      }
+      if (mInput.maxTFsPerFile > 0) {
+        entries = std::min(entries, mInput.maxTFsPerFile);
+      }
+      mInput.ctfIDs.clear();
+      mInput.ctfIDs.resize(entries);
+      std::iota(mInput.ctfIDs.begin(), mInput.ctfIDs.end(), 0);
+      std::shuffle(mInput.ctfIDs.begin(), mInput.ctfIDs.end(), gen);
+      LOGP(info, "will shuffle CTF entries in this order:");
+      for (int i{0}; i < (int)mInput.ctfIDs.size(); ++i) {
+        LOGP(info, "\tTF {:02} -> {:02}", i, mInput.ctfIDs[i]);
+      }
+    }
   } catch (const std::exception& e) {
     LOG(error) << "Cannot process " << flname << ", reason: " << e.what();
     mCTFTree.reset();
@@ -322,9 +345,12 @@ void CTFReaderSpec::run(ProcessingContext& pc)
   long startWait = 0;
 
   while (mRunning) {
-    if (mCTFTree) { // there is a tree open with multiple CTF
-      if (mInput.ctfIDs.empty() || mInput.ctfIDs[mSelIDEntry] == mCTFCounter) { // no selection requested or matching CTF ID is found
+    if (mCTFTree) {                                                                               // there is a tree open with multiple CTF
+      if (mInput.ctfIDs.empty() || mInput.ctfIDs[mSelIDEntry] == mCTFCounter || mInput.shuffle) { // no selection requested or matching CTF ID is found
         LOG(debug) << "TF " << mCTFCounter << " of " << mInput.maxTFs << " loop " << mFileFetcher->getNLoops();
+        if (mInput.shuffle) {
+          mCurrTreeEntry = mInput.ctfIDs[mSelIDEntry];
+        }
         mSelIDEntry++;
         if (processTF(pc)) {
           break;
@@ -501,7 +527,13 @@ bool CTFReaderSpec::processTF(ProcessingContext& pc)
 void CTFReaderSpec::checkTreeEntries()
 {
   // check if the tree has entries left, if needed, close current tree/file
-  if (++mCurrTreeEntry >= mCTFTree->GetEntries() || (mInput.maxTFsPerFile > 0 && mCurrTreeEntry >= mInput.maxTFsPerFile)) { // this file is done, check if there are other files
+  bool reachedEnd{false};
+  if (mInput.shuffle) { // last entry is last id
+    reachedEnd = (mCurrTreeEntry == mInput.ctfIDs.back());
+  } else {
+    reachedEnd = (mCurrTreeEntry++ >= mCTFTree->GetEntries());
+  }
+  if (reachedEnd || (mInput.maxTFsPerFile > 0 && mCurrTreeEntry >= mInput.maxTFsPerFile)) { // this file is done, check if there are other files
     mCTFTree.reset();
     mCTFFile->Close();
     mCTFFile.reset();

Detectors/CTF/workflow/src/ctf-reader-workflow.cxx

@@ -56,6 +56,7 @@ void customize(std::vector<o2::framework::ConfigParamSpec>& workflowOptions)
   options.push_back(ConfigParamSpec{"skipDet", VariantType::String, std::string{DetID::NONE}, {"comma-separate list of detectors to skip"}});
   options.push_back(ConfigParamSpec{"loop", VariantType::Int, 0, {"loop N times (infinite for N<0)"}});
   options.push_back(ConfigParamSpec{"delay", VariantType::Float, 0.f, {"delay in seconds between consecutive TFs sending"}});
+  options.push_back(ConfigParamSpec{"shuffle", VariantType::Bool, false, {"shuffle TF sending order (for debug)"}});
   options.push_back(ConfigParamSpec{"copy-cmd", VariantType::String, "alien_cp ?src file://?dst", {"copy command for remote files or no-copy to avoid copying"}}); // Use "XrdSecPROTOCOL=sss,unix xrdcp -N root://eosaliceo2.cern.ch/?src ?dst" for direct EOS access
   options.push_back(ConfigParamSpec{"ctf-file-regex", VariantType::String, ".*o2_ctf_run.+\\.root$", {"regex string to identify CTF files"}});
   options.push_back(ConfigParamSpec{"remote-regex", VariantType::String, "^(alien://|)/alice/data/.+", {"regex string to identify remote files"}}); // Use "^/eos/aliceo2/.+" for direct EOS access
@@ -120,6 +121,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& configcontext)
 
   ctfInput.maxFileCache = std::max(1, configcontext.options().get<int>("max-cached-files"));
 
+  ctfInput.shuffle = configcontext.options().get<bool>("shuffle");
   ctfInput.copyCmd = configcontext.options().get<std::string>("copy-cmd");
   ctfInput.tffileRegex = configcontext.options().get<std::string>("ctf-file-regex");
   ctfInput.remoteRegex = configcontext.options().get<std::string>("remote-regex");