HybridGenerator: Asyncronous + parallel event generation (#13788)

This introduces:
* asyncronous event generation
* possibility for parallel event generation

This is useful for:
* hiding latency (IO) of certain generators
* decoupling the actual work from the call sequence into HybridGenerator
* collaboration from multiple clones of the same generator to generate
  a certain number of events

The implementation relies on tbb::task_arena and input/output queues
for decoupling the task_arena from the HybridGenerator thread.

An example is added for parallel event gen.

In addition, this commit has the following changes:

- Small adjustments to seeding of Pythia8 in order
  to avoid same seeds in multiple parallel Pythia instances.

- possibility to init external generator from an INI file (typically
  done or available in O2DPG)

- use shared_ptr instead of unique_ptr to keep generators in order
  to avoid lifetime problems with async processing

- preparion to run underlying generator Init() functions in async way
  (not active yet; needs checks if generators are thread safe)

(cherry picked from commit f4d9b9c)

Author: sawenzel

Generators/include/Generators/Generator.h

@@ -141,6 +141,11 @@ class Generator : public FairGenerator
   /** lorentz boost data members **/
   Double_t mBoost;
 
+  // a unique generator instance counter
+  // this can be used to make sure no two generator instances have the same seed etc.
+  static std::atomic<int> InstanceCounter;
+  int mThisInstanceID = 0;
+
  private:
   void updateSubGeneratorInformation(o2::dataformats::MCEventHeader* header) const;
 

Generators/include/Generators/GeneratorExternalParam.h

@@ -37,6 +37,7 @@ struct GeneratorExternalParam : public o2::conf::ConfigurableParamHelper<Generat
 struct ExternalGenConfig {
   std::string fileName = "";
   std::string funcName = "";
+  std::string iniFile = ""; // if ini file is given, the configuration will be taken from this and the other 2 fields neglected
 };
 
 } // end namespace eventgen

Generators/include/Generators/GeneratorHybrid.h

@@ -39,6 +39,12 @@
 #include <rapidjson/writer.h>
 #include "TBufferJSON.h"
 
+#include <tbb/concurrent_queue.h>
+#include <tbb/task_arena.h>
+#include <iostream>
+#include <thread>
+#include <atomic>
+
 namespace o2
 {
 namespace eventgen
@@ -50,25 +56,29 @@ class GeneratorHybrid : public Generator
  public:
   GeneratorHybrid() = default;
   GeneratorHybrid(const std::string& inputgens);
-  ~GeneratorHybrid() = default;
+  ~GeneratorHybrid();
 
   Bool_t Init() override;
   Bool_t generateEvent() override;
   Bool_t importParticles() override;
 
+  void setNEvents(int n) { mNEvents = n; }
+
   Bool_t parseJSON(const std::string& path);
   template <typename T>
   std::string jsonValueToString(const T& value);
 
  private:
   o2::eventgen::Generator* currentgen = nullptr;
-  std::vector<std::unique_ptr<o2::eventgen::Generator>> gens;
+  std::vector<std::shared_ptr<o2::eventgen::Generator>> gens;
   const std::vector<std::string> generatorNames = {"extkinO2", "evtpool", "boxgen", "external", "hepmc", "pythia8", "pythia8pp", "pythia8hi", "pythia8hf", "pythia8powheg"};
   std::vector<std::string> mInputGens;
   std::vector<std::string> mGens;
   std::vector<std::string> mConfigs;
   std::vector<std::string> mConfsPythia8;
 
+  std::vector<bool> mGenIsInitialized;
+
   // Parameters configurations
   std::vector<std::unique_ptr<o2::eventgen::BoxGenConfig>> mBoxGenConfigs;
   std::vector<std::unique_ptr<o2::eventgen::Pythia8GenConfig>> mPythia8GenConfigs;
@@ -84,6 +94,28 @@ class GeneratorHybrid : public Generator
   int mseqCounter = 0;
   int mCurrentFraction = 0;
   int mIndex = 0;
+  int mEventCounter = 0;
+  int mTasksStarted = 0;
+
+  // Create a task arena with a specified number of threads
+  std::thread mTBBTaskPoolRunner;
+  tbb::concurrent_bounded_queue<int> mInputTaskQueue;
+  std::vector<tbb::concurrent_bounded_queue<int>> mResultQueue;
+  tbb::task_arena mTaskArena;
+  std::atomic<bool> mStopFlag;
+  bool mIsInitialized = false;
+
+  int mNEvents = -1; // the number of events to be done, if known (helps initiating cleanup)
+
+  enum class GenMode {
+    kSeq,
+    kParallel
+  };
+
+  // hybrid gen operation mode - should be either 'sequential' or 'parallel'
+  // parallel means that we have clones of the same generator collaborating on event generation
+  // sequential means that events will be produced in the order given by fractions; async processing is still happening
+  GenMode mGenerationMode = GenMode::kSeq; //!
 };
 
 } // namespace eventgen

Generators/include/Generators/GeneratorHybridParam.h

@@ -31,6 +31,7 @@ namespace eventgen
 struct GeneratorHybridParam : public o2::conf::ConfigurableParamHelper<GeneratorHybridParam> {
   std::string configFile = ""; // JSON configuration file for the generators
   bool randomize = false;      // randomize the order of the generators, if not generator using fractions
+  int num_workers = 1;         // number of threads available for asyn/parallel event generation
   O2ParamDef(GeneratorHybridParam, "GeneratorHybrid");
 };
 

Generators/include/Generators/GeneratorPythia8.h

@@ -287,6 +287,9 @@ class GeneratorPythia8 : public Generator
                                // Value of -1 means unitialized; 0 will be time-dependent and values >1 <= MAX_SEED concrete reproducible seeding
   Pythia8GenConfig mGenConfig; // configuration object
 
+  static std::atomic<int> Pythia8InstanceCounter;
+  int mThisPythia8InstanceID = 0;
+
   constexpr static long MAX_SEED = 900000000;
 
   ClassDefOverride(GeneratorPythia8, 1);

Generators/src/Generator.cxx

@@ -28,13 +28,17 @@ namespace o2
 namespace eventgen
 {
 
+std::atomic<int> Generator::InstanceCounter{0};
+
 /*****************************************************************/
 /*****************************************************************/
 
 Generator::Generator() : FairGenerator("ALICEo2", "ALICEo2 Generator"),
                          mBoost(0.)
 {
   /** default constructor **/
+  mThisInstanceID = Generator::InstanceCounter;
+  Generator::InstanceCounter++;
 }
 
 /*****************************************************************/
@@ -43,6 +47,8 @@ Generator::Generator(const Char_t* name, const Char_t* title) : FairGenerator(na
                                                                 mBoost(0.)
 {
   /** constructor **/
+  mThisInstanceID = Generator::InstanceCounter;
+  Generator::InstanceCounter++;
 }
 
 /*****************************************************************/

Generators/src/GeneratorFactory.cxx

@@ -285,6 +285,7 @@ void GeneratorFactory::setPrimaryGenerator(o2::conf::SimConfig const& conf, Fair
       return;
     }
     auto hybrid = new o2::eventgen::GeneratorHybrid(config);
+    hybrid->setNEvents(conf.getNEvents());
     primGen->AddGenerator(hybrid);
 #endif
   } else {