Integrated code review by Sandro + improvements

Author: jackal1-66

Generators/CMakeLists.txt

@@ -152,6 +152,14 @@ if(doBuildSimulation)
                 PUBLIC_LINK_LIBRARIES O2::Generators)
   endif()
 
+  if(pythia_FOUND AND HepMC3_FOUND)
+    o2_add_test(GeneratorHybridEventPool NAME test_Generator_test_GeneratorHybridEventPool
+                SOURCES test/test_GeneratorHybridEventPool.cxx
+                COMPONENT_NAME Generator
+                LABELS generator
+                PUBLIC_LINK_LIBRARIES O2::Generators)
+  endif()
+
   # o2_add_test(GeneratorPythia8Param NAME test_Generator_test_GeneratorPythia8Param
   #             SOURCES test/test_GeneratorPythia8Param.cxx
   #             COMPONENT_NAME Generator

Generators/include/Generators/Generator.h

@@ -18,6 +18,7 @@
 #include "TParticle.h"
 #include "Generators/Trigger.h"
 #include <functional>
+#include <optional>
 #include <vector>
 #include <unordered_map>
 
@@ -94,10 +95,13 @@ class Generator : public FairGenerator
   void addDeepTrigger(DeepTrigger trigger) { mDeepTriggers.push_back(trigger); };
   // setter for global number of events
   static void setTotalNEvents(unsigned int& n) { gTotalNEvents = n; }
+  // number of events this generator instance is expected to serve; 0 means unknown
+  void setExpectedNEvents(unsigned int n) { mExpectedNEvents = n; }
 
   /** getters **/
   const std::vector<TParticle>& getParticles() const { return mParticles; }; //!
   static unsigned int getTotalNEvents() { return gTotalNEvents; };
+  unsigned int getExpectedNEvents() const { return mExpectedNEvents.value_or(gTotalNEvents); }
 
   /** other **/
   void clearParticles() { mParticles.clear(); };
@@ -170,6 +174,8 @@ class Generator : public FairGenerator
 
   // global static information about (upper limit of) number of events to be generated
   static unsigned int gTotalNEvents;
+  // number of events expected to be served by this instance. Number of events of the job if unset
+  std::optional<unsigned int> mExpectedNEvents; //!
 
   // Loopers generator instance
   o2::eventgen::GenTPCLoopers* mTPCLoopersGen = nullptr;

Generators/include/Generators/GeneratorHybrid.h

@@ -44,6 +44,7 @@
 #include <iostream>
 #include <thread>
 #include <atomic>
+#include <exception>
 
 namespace o2
 {
@@ -74,6 +75,12 @@ class GeneratorHybrid : public Generator
  private:
   GeneratorHybrid(const std::string& inputgens);
   ~GeneratorHybrid();
+  // number of events each sub-generator is expected to serve (0 == unknown)
+  std::vector<unsigned int> expectedNEventsPerGenerator(unsigned int nevents) const;
+  // schedules the generation of an event by a sub-generator, within its request budget
+  void requestEvent(int genIndex);
+  // re-raises, in the calling thread, a failure of a sub-generator during the generation of its event
+  void rethrowSubGeneratorFailure(int genIndex);
   o2::eventgen::Generator* currentgen = nullptr;
   std::vector<std::shared_ptr<o2::eventgen::Generator>> gens;
   const std::vector<std::string> generatorNames = {"evtpool", "boxgen", "external", "hepmc", "pythia8", "pythia8pp", "pythia8hi", "pythia8hf", "pythia8powheg"};
@@ -101,6 +108,9 @@ class GeneratorHybrid : public Generator
   int mIndex = 0;
   int mEventCounter = 0;
   int mTasksStarted = 0;
+  // number of events which are requested from each sub-generator. This is used if the share of
+  // each sub-generator is known exactly (fixed fractions, sequential or cocktail mode), empty otherwise
+  std::vector<unsigned int> mRequestBudget;
 
   // Cocktail mode
   bool mCocktailMode = false;
@@ -117,6 +127,8 @@ class GeneratorHybrid : public Generator
   std::vector<tbb::concurrent_bounded_queue<int>> mResultQueue;
   tbb::task_arena mTaskArena;
   std::atomic<bool> mStopFlag;
+  // failure of the last event generation of each sub-generator in a worker thread
+  std::vector<std::exception_ptr> mGenFailures;
   bool mIsInitialized = false;
 
   o2::dataformats::MCEventHeader mMCEventHeader; // to capture event headers

Generators/src/GeneratorFromFile.cxx

@@ -217,6 +217,8 @@ void GeneratorFromO2Kine::closeCurrentFile()
   mEventOrder.clear();
   mEventsAvailable = 0;
   mEventCounter = 0;
+  // the header read from the previous file must not leak into events of the next one
+  mOrigMCEventHeader.reset();
   if (mCurrentFile) {
     mCurrentFile->Close();
     delete mCurrentFile;
@@ -242,44 +244,46 @@ bool GeneratorFromO2Kine::openFile(int index)
 {
   // opens one file of the list, connects the branches to it and fixes the
   // order in which its events are going to be served;
-  // any previously open file is closed first, so that we never keep more than
-  // one input file open at a time
-  closeCurrentFile();
+  // the file which is currently open is only closed once the new one has been found usable
+  // In the standard case never more than one input file is open at a time
   if (index < 0 || index >= (int)mFileNames.size()) {
     return false;
   }
   auto const& name = mFileNames[index];
 
-  mCurrentFile = TFile::Open(name.c_str());
-  if (mCurrentFile == nullptr || mCurrentFile->IsZombie()) {
+  std::unique_ptr<TFile> file(TFile::Open(name.c_str()));
+  if (file == nullptr || file->IsZombie()) {
     LOG(error) << "EventFile " << name << " could not be opened";
-    closeCurrentFile();
     return false;
   }
   // the kinematics will be stored inside a branch MCTrack
   // different events are stored inside different entries
-  auto tree = (TTree*)mCurrentFile->Get("o2sim");
+  auto tree = (TTree*)file->Get("o2sim");
   if (!tree) {
     LOG(error) << "EventFile " << name << " does not contain an 'o2sim' tree";
-    closeCurrentFile();
     return false;
   }
-  mEventBranch = tree->GetBranch("MCTrack");
-  if (!mEventBranch) {
+  auto eventBranch = tree->GetBranch("MCTrack");
+  if (!eventBranch) {
     LOG(error) << "No MCTrack branch found in " << name;
-    closeCurrentFile();
     return false;
   }
-  mEventsAvailable = mEventBranch->GetEntries();
-  if (mEventsAvailable <= 0) {
+  auto eventsAvailable = eventBranch->GetEntries();
+  if (eventsAvailable <= 0) {
     LOG(warn) << "EventFile " << name << " does not contain any event";
-    closeCurrentFile();
     return false;
   }
-  mMCHeaderBranch = tree->GetBranch("MCEventHeader.");
-  if (!mMCHeaderBranch) {
+  auto mcHeaderBranch = tree->GetBranch("MCEventHeader.");
+  if (!mcHeaderBranch) {
     LOG(warn) << "No MCEventHeader branch found in kinematics input file";
   }
+
+  // the new file is good: release the previous one and take over
+  closeCurrentFile();
+  mCurrentFile = file.release();
+  mEventBranch = eventBranch;
+  mEventsAvailable = eventsAvailable;
+  mMCHeaderBranch = mcHeaderBranch;
   establishEventOrder();
   mCurrentFileIndex = index;
   mEventCounter = 0;
@@ -303,7 +307,7 @@ bool GeneratorFromO2Kine::openNextFile(bool wrapAround)
       if (!wrapAround) {
         return false;
       }
-      if (trial == 0) {
+      if (next == numFiles) {
         LOG(info) << "Reached the end of the input file list; reusing events from the beginning";
       }
       next = next % numFiles;
@@ -380,19 +384,22 @@ bool GeneratorFromO2Kine::Init()
   }
   mCurrentFileIndex = -1;
   if (!openNextFile(false)) {
-    LOG(error) << "Problem reading events from the given kinematics input";
+    LOG(fatal) << "Problem reading events from the given kinematics input: none of the " << mFileNames.size()
+               << " input file(s) can be used";
     return false;
   }
   if (mStartEvent > 0) {
-    if (mStartEvent < mEventsAvailable) {
+    if (mRandomize) {
+      LOG(warn) << "Start event " << mStartEvent << " ignored: the events are served in random order (randomize)";
+    } else if (mStartEvent < mEventsAvailable) {
       mEventCounter = mStartEvent;
     } else {
       LOG(error) << "start event bigger than available events";
     }
   }
   // Simple estimate of events without checking all the files.
   // To be discussed if we want instead to do this, or provide an additional file with the pools
-  auto requested = getTotalNEvents();
+  auto requested = getExpectedNEvents();
   if (requested > 0 && !mRoundRobin && mEventsAvailable > 0) {
     auto estimate = (size_t)mEventsAvailable * mFileNames.size();
     if (estimate < requested) {
@@ -422,7 +429,7 @@ bool GeneratorFromO2Kine::importParticles()
   // Next file in the list opened when the events of the current one are used up
   if (mEventCounter >= mEventsAvailable) {
     if (!openNextFile(mRoundRobin)) {
-      auto requested = getTotalNEvents();
+      auto requested = getExpectedNEvents();
       LOG(fatal) << "GeneratorFromO2Kine: ran out of events after " << mEventsServed
                  << " event(s) from " << mFilesUsed << " input file(s)"
                  << (requested > 0 ? " (" + std::to_string(requested) + " were requested)" : "")
@@ -584,7 +591,7 @@ bool GeneratorFromEventPool::Init()
   mPoolFilesAvailable = setupFileUniverse(expPath.Data());
 
   if (mPoolFilesAvailable.size() == 0) {
-    LOG(error) << "No file found that can be used with EventPool generator";
+    LOG(fatal) << "No file found that can be used with EventPool generator (eventPoolPath: " << mConfig.eventPoolPath << ")";
     return false;
   }
   LOG(info) << "Found " << mPoolFilesAvailable.size() << " available event pool files";
@@ -602,14 +609,18 @@ bool GeneratorFromEventPool::Init()
     .rngseed = mConfig.rngseed,
     .randomphi = mConfig.randomphi};
   mO2KineGenerator.reset(new GeneratorFromO2Kine(kine_config, mFilesChosen));
+  // the internal generator serves the events of this one
+  mO2KineGenerator->setExpectedNEvents(getExpectedNEvents());
   return mO2KineGenerator->Init();
 }
 
 std::vector<std::string> GeneratorFromEventPool::selectFiles(std::vector<std::string> const& universe)
 {
   // shuffles the whole pool universe so that different jobs go through it in a
-  // different order
+  // different order. Sorting first as a directory listing (or alien find) gives no fixed order,
+  // and the same rngseed must give the same order everywhere
   auto result = universe;
+  std::sort(result.begin(), result.end());
   std::shuffle(result.begin(), result.end(), mRandomEngine);
   return result;
 }

Generators/src/GeneratorHepMC.cxx

@@ -706,7 +706,7 @@ Bool_t GeneratorHepMC::generateEventOrdered()
   // The entry to be read is fixed by the event order established at file opening
   if (mEventCounter >= (int)mEventOrder.size()) {
     if (not mRoundRobin) {
-      auto requested = getTotalNEvents();
+      auto requested = getExpectedNEvents();
       LOG(fatal) << "GeneratorHepMC: ran out of events after " << mEventsServed
                  << " event(s) from " << mCurrentFileName
                  << (requested > 0 ? " (" + std::to_string(requested) + " were requested)" : "")
@@ -863,7 +863,7 @@ Bool_t GeneratorHepMC::Init()
       return false;
     }
     establishEventOrder();
-    auto requested = getTotalNEvents();
+    auto requested = getExpectedNEvents();
     if (requested > 0 and not mRoundRobin and mEventOrder.size() < requested) {
       LOG(warn) << "This job will request " << requested << " events, but the input holds "
                 << "only " << mEventOrder.size() << " usable event(s). The job will stop "