diff --git a/Generators/CMakeLists.txt b/Generators/CMakeLists.txt index 094510c2c61a5..5415327bddb8b 100644 --- a/Generators/CMakeLists.txt +++ b/Generators/CMakeLists.txt @@ -139,6 +139,11 @@ if(doBuildSimulation) LABELS generator PUBLIC_LINK_LIBRARIES O2::Generators) + o2_add_test(EventPoolChaining NAME test_Generator_test_EventPoolChaining + SOURCES test/test_EventPoolChaining.cxx + COMPONENT_NAME Generator + LABELS generator + PUBLIC_LINK_LIBRARIES O2::Generators) if(HepMC3_FOUND) o2_add_test(GeneratorHepMCIndexed NAME test_Generator_test_GeneratorHepMCIndexed SOURCES test/test_GeneratorHepMCIndexed.cxx @@ -147,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 diff --git a/Generators/include/Generators/Generator.h b/Generators/include/Generators/Generator.h index 7001a48410dd7..8c5ee2b7df930 100644 --- a/Generators/include/Generators/Generator.h +++ b/Generators/include/Generators/Generator.h @@ -18,6 +18,7 @@ #include "TParticle.h" #include "Generators/Trigger.h" #include +#include #include #include @@ -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& 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 mExpectedNEvents; //! // Loopers generator instance o2::eventgen::GenTPCLoopers* mTPCLoopersGen = nullptr; diff --git a/Generators/include/Generators/GeneratorFromFile.h b/Generators/include/Generators/GeneratorFromFile.h index ad93814467fc3..d2e764c3d001c 100644 --- a/Generators/include/Generators/GeneratorFromFile.h +++ b/Generators/include/Generators/GeneratorFromFile.h @@ -21,6 +21,8 @@ #include #include #include +#include +#include class TBranch; class TFile; @@ -66,13 +68,18 @@ class GeneratorFromFile : public FairGenerator }; /// This class implements a generic FairGenerator which -/// reads the particles from an external O2 sim kinematics file. +/// reads the particles from one or more external O2 sim kinematics files. class GeneratorFromO2Kine : public o2::eventgen::Generator { public: GeneratorFromO2Kine() = default; + /// name may be a single file or a comma-separated list of files to be read one after the other GeneratorFromO2Kine(const char* name); + GeneratorFromO2Kine(std::vector const& filenames); GeneratorFromO2Kine(O2KineGenConfig const& pars); + /// same as above but with an explicit list of files. Used for event pools + GeneratorFromO2Kine(O2KineGenConfig const& pars, std::vector const& filenames); + ~GeneratorFromO2Kine() override; bool Init() override; @@ -91,24 +98,58 @@ class GeneratorFromO2Kine : public o2::eventgen::Generator void updateHeader(o2::dataformats::MCEventHeader* eventHeader) override; const o2::dataformats::MCEventHeader* getOrigMCEventHeader() const { return mOrigMCEventHeader.get(); } + /// number of events available in the file that is currently open + int getEventsAvailable() const { return mEventsAvailable; } + /// number of input files this generator can go through + int getNumberOfFiles() const { return (int)mFileNames.size(); } + /// index (within the file list) of the file currently open, -1 if none + int getCurrentFileIndex() const { return mCurrentFileIndex; } + /// name of the file currently open, empty if none + std::string getCurrentFileName() const; + /// number of opened files so far (including the current one) + int getNumberOfFilesUsed() const { return mFilesUsed; } + /// total number of events delivered so far + int getEventsServed() const { return mEventsServed; } + + /// helper splitting a comma-separated list of file names into its components + static std::vector splitFileNames(std::string const& filenames); + private: - TFile* mEventFile = nullptr; //! the file containing the persistent events - TBranch* mEventBranch = nullptr; //! the branch containing the persistent events - TBranch* mMCHeaderBranch = nullptr; //! branch containing MC event headers - int mEventCounter = 0; - int mEventsAvailable = 0; - bool mSkipNonTrackable = true; //! whether to pass non-trackable (decayed particles) to the MC stack - bool mContinueMode = false; //! whether we want to continue simulation of previously inhibited tracks - bool mRoundRobin = false; //! whether we want to take events from file in a round robin fashion - bool mRandomize = false; //! whether we want to randomize the order of events in the input file - unsigned int mRngSeed = 0; //! randomizer seed, 0 for random value - bool mRandomPhi = false; //! whether we want to randomize the phi angle of the particles - TGrid* mAlienInstance = nullptr; // a cached connection to TGrid (needed for Alien locations) + /// closes the currently opened file currently + void closeCurrentFile(); + /// fixes the order in which the events of the current file are served + void establishEventOrder(); + /// opens the file at the given index of the file list. + /// Returns false in case the file cannot be used + bool openFile(int index); + /// moves on to the next usable file of the list; wraps around in round robin mode. + /// Returns false when no further file is available + bool openNextFile(bool wrapAround); + + std::vector mFileNames; //! the list of input files, read one after the other + int mCurrentFileIndex = -1; //! index of the file currently open + int mFilesUsed = 0; //! how many files have been opened so far + TFile* mCurrentFile = nullptr; //! the file currently open + TBranch* mEventBranch = nullptr; //! the branch containing the persistent events + TBranch* mMCHeaderBranch = nullptr; //! branch containing MC event headers + std::vector mEventOrder; //! order in which the entries of the current file are served + int mEventCounter = 0; //! events already delivered from the current file + int mEventsServed = 0; //! events delivered in total, across all files + int mEventsAvailable = 0; //! events contained in the current file + int mStartEvent = 0; //! event to start from in the very first file + int mLastEntryRead = -1; //! entry of the current event within the current file + bool mSkipNonTrackable = true; //! whether to pass non-trackable (decayed particles) to the MC stack + bool mContinueMode = false; //! whether we want to continue simulation of previously inhibited tracks + bool mRoundRobin = false; //! whether we want to take events from file in a round robin fashion + bool mRandomize = false; //! whether we want to randomize the order of events in the input file + unsigned int mRngSeed = 0; //! randomizer seed, 0 for random value + bool mRandomPhi = false; //! whether we want to randomize the phi angle of the particles + TGrid* mAlienInstance = nullptr; // a cached connection to TGrid (needed for Alien locations) std::unique_ptr mConfig; //! Configuration object std::unique_ptr mOrigMCEventHeader; //! the MC event header of the original file - ClassDefOverride(GeneratorFromO2Kine, 2); + ClassDefOverride(GeneratorFromO2Kine, 3); }; /// Special generator for event pools. @@ -167,15 +208,24 @@ class GeneratorFromEventPool : public o2::eventgen::Generator std::vector const& getFileUniverse() const { return mPoolFilesAvailable; } + /// the file universe, in the order this generator instance will go through it + std::vector const& getChosenFiles() const { return mFilesChosen; } + + /// shuffles the given universe of pool files into the order this instance will use + std::vector selectFiles(std::vector const& universe); + + /// access to the underlying kinematics generator + o2::eventgen::GeneratorFromO2Kine const* getO2KineGenerator() const { return mO2KineGenerator.get(); } + private: EventPoolGenConfig mConfig; //! Configuration object std::unique_ptr mO2KineGenerator = nullptr; //! actual generator doing the work std::vector mPoolFilesAvailable; //! container keeping the collection of files in the event pool - std::string mFileChosen; //! the file chosen for the pool + std::vector mFilesChosen; //! the file(s) chosen from the pool // random number generator to determine a concrete file name std::mt19937 mRandomEngine; //! - ClassDefOverride(GeneratorFromEventPool, 1); + ClassDefOverride(GeneratorFromEventPool, 2); }; } // end namespace eventgen diff --git a/Generators/include/Generators/GeneratorFromO2KineParam.h b/Generators/include/Generators/GeneratorFromO2KineParam.h index e8d886186e2d2..5bc369291c29d 100644 --- a/Generators/include/Generators/GeneratorFromO2KineParam.h +++ b/Generators/include/Generators/GeneratorFromO2KineParam.h @@ -31,22 +31,24 @@ namespace eventgen struct GeneratorFromO2KineParam : public o2::conf::ConfigurableParamHelper { bool skipNonTrackable = true; bool continueMode = false; - bool roundRobin = false; // read events with period boundary conditions - bool randomize = false; // randomize the order of events + bool roundRobin = false; // start over from the first file/event once all events have been used + bool randomize = false; // serve the events of each file in random order (each one exactly once) unsigned int rngseed = 0; // randomizer seed, 0 for random value bool randomphi = false; // randomize phi angle - std::string fileName = ""; // filename to read from - takes precedence over SimConfig if given + std::string fileName = ""; // filename(s) to read from - takes precedence over SimConfig if given; + // a comma-separated list of files is read one file after the other O2ParamDef(GeneratorFromO2KineParam, "GeneratorFromO2Kine"); }; struct O2KineGenConfig { bool skipNonTrackable = true; bool continueMode = false; - bool roundRobin = false; // read events with period boundary conditions - bool randomize = false; // randomize the order of events + bool roundRobin = false; // start over from the first file/event once all events have been used + bool randomize = false; // serve the events of each file in random order (each one exactly once) unsigned int rngseed = 0; // randomizer seed, 0 for random value bool randomphi = false; // randomize phi angle - std::string fileName = ""; // filename to read from - takes precedence over SimConfig if given + std::string fileName = ""; // filename(s) to read from - takes precedence over SimConfig if given; + // a comma-separated list of files is read one file after the other }; struct EventPoolGenConfig { @@ -54,8 +56,8 @@ struct EventPoolGenConfig { // or .. a local file containing a list of files to use // or .. a concrete file path to a kinematics file bool skipNonTrackable = true; // <--- do we need this? - bool roundRobin = false; // read events with period boundary conditions - bool randomize = true; // randomize the order of events + bool roundRobin = false; // start over from the first file/event once all events have been used + bool randomize = true; // serve the events of each file in random order (each one exactly once) unsigned int rngseed = 0; // randomizer seed, 0 for random value bool randomphi = false; // randomize phi angle; rotates tracks in events by some phi-angle }; diff --git a/Generators/include/Generators/GeneratorHybrid.h b/Generators/include/Generators/GeneratorHybrid.h index b57257747c629..a7eea02db28b2 100644 --- a/Generators/include/Generators/GeneratorHybrid.h +++ b/Generators/include/Generators/GeneratorHybrid.h @@ -44,6 +44,7 @@ #include #include #include +#include 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 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> gens; const std::vector 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 mRequestBudget; // Cocktail mode bool mCocktailMode = false; @@ -117,6 +127,8 @@ class GeneratorHybrid : public Generator std::vector> mResultQueue; tbb::task_arena mTaskArena; std::atomic mStopFlag; + // failure of the last event generation of each sub-generator in a worker thread + std::vector mGenFailures; bool mIsInitialized = false; o2::dataformats::MCEventHeader mMCEventHeader; // to capture event headers diff --git a/Generators/src/GeneratorFromFile.cxx b/Generators/src/GeneratorFromFile.cxx index 787e315ce80ca..88487aa94d7a9 100644 --- a/Generators/src/GeneratorFromFile.cxx +++ b/Generators/src/GeneratorFromFile.cxx @@ -13,6 +13,7 @@ #include "Generators/GeneratorFromO2KineParam.h" #include "SimulationDataFormat/MCTrack.h" #include "SimulationDataFormat/MCEventHeader.h" +#include "CommonUtils/StringUtils.h" #include #include #include @@ -21,6 +22,10 @@ #include #include #include +#include +#include +#include +#include #include #include #include @@ -170,77 +175,249 @@ Bool_t GeneratorFromFile::ReadEvent(FairPrimaryGenerator* primGen) // based on O2 kinematics -GeneratorFromO2Kine::GeneratorFromO2Kine(const char* name) +namespace { - // this generator should leave all dimensions the same as in the incoming kinematics file - setMomentumUnit(1.); - setEnergyUnit(1.); - setPositionUnit(1.); - setTimeUnit(1.); - - if (strncmp(name, "alien:/", 7) == 0 && !gGrid) { - TGrid::Connect("alien:"); - if (!gGrid) { - LOG(fatal) << "Could not connect to alien, did you check the alien token?"; +// connects to AliEn in case at least one of the given file names lives there +void connectToAlienIfNeeded(std::vector const& filenames) +{ + if (gGrid) { + return; + } + for (auto const& name : filenames) { + if (name.starts_with("alien:/")) { + TGrid::Connect("alien:"); + if (!gGrid) { + LOG(fatal) << "Could not connect to alien, did you check the alien token?"; + } return; } } - mEventFile = TFile::Open(name); - if (mEventFile == nullptr) { - LOG(fatal) << "EventFile " << name << " not found"; - return; +} +} // namespace + +std::vector GeneratorFromO2Kine::splitFileNames(std::string const& filenames) +{ + // splits a comma-separated list of file names, trimming white space and dropping empty tokens + return o2::utils::Str::tokenize(filenames, ','); +} + +std::string GeneratorFromO2Kine::getCurrentFileName() const +{ + if (mCurrentFileIndex < 0 || mCurrentFileIndex >= (int)mFileNames.size()) { + return std::string(); + } + return mFileNames[mCurrentFileIndex]; +} + +void GeneratorFromO2Kine::closeCurrentFile() +{ + // the branches belong to the tree of the file, so they die with it + mEventBranch = nullptr; + mMCHeaderBranch = nullptr; + 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; + mCurrentFile = nullptr; + } +} + +void GeneratorFromO2Kine::establishEventOrder() +{ + // The order in which the events of the current file are served is decided here + mEventOrder.resize(mEventsAvailable); + std::iota(mEventOrder.begin(), mEventOrder.end(), 0); + if (mRandomize) { + // Fisher-Yates shuffle based on the ROOT random generator + for (int i = mEventsAvailable - 1; i > 0; --i) { + auto j = (int)gRandom->Integer(i + 1); + std::swap(mEventOrder[i], mEventOrder[j]); + } + } +} + +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; + // 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]; + + std::unique_ptr file(TFile::Open(name.c_str())); + if (file == nullptr || file->IsZombie()) { + LOG(error) << "EventFile " << name << " could not be opened"; + return false; } // the kinematics will be stored inside a branch MCTrack // different events are stored inside different entries - auto tree = (TTree*)mEventFile->Get("o2sim"); - if (tree) { - mEventBranch = tree->GetBranch("MCTrack"); - if (mEventBranch) { - mEventsAvailable = mEventBranch->GetEntries(); - LOG(info) << "Found " << mEventsAvailable << " events in this file"; + auto tree = (TTree*)file->Get("o2sim"); + if (!tree) { + LOG(error) << "EventFile " << name << " does not contain an 'o2sim' tree"; + return false; + } + auto eventBranch = tree->GetBranch("MCTrack"); + if (!eventBranch) { + LOG(error) << "No MCTrack branch found in " << name; + return false; + } + auto eventsAvailable = eventBranch->GetEntries(); + if (eventsAvailable <= 0) { + LOG(warn) << "EventFile " << name << " does not contain any event"; + return false; + } + 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; + mFilesUsed++; + LOG(info) << "Reading events from kinematics file " << name << " (" << mEventsAvailable + << " events, " << (mRandomize ? "randomized" : "sequential") << " order)"; + return true; +} + +bool GeneratorFromO2Kine::openNextFile(bool wrapAround) +{ + // advances to the next usable file of the list; unusable files are skipped. + auto numFiles = (int)mFileNames.size(); + if (numFiles == 0) { + return false; + } + // we try each of the remaining files at most once + for (int trial = 0; trial < numFiles; ++trial) { + auto next = mCurrentFileIndex + 1 + trial; + if (next >= numFiles) { + if (!wrapAround) { + return false; + } + if (next == numFiles) { + LOG(info) << "Reached the end of the input file list; reusing events from the beginning"; + } + next = next % numFiles; } - mMCHeaderBranch = tree->GetBranch("MCEventHeader."); - if (mMCHeaderBranch) { - LOG(info) << "Found " << mMCHeaderBranch->GetEntries() << " event-headers"; - } else { - LOG(warn) << "No MCEventHeader branch found in kinematics input file"; + if (next == mCurrentFileIndex && mCurrentFile) { + // this is the only usable file and it is already open - restart from it, + // drawing a fresh event order + establishEventOrder(); + mEventCounter = 0; + return true; + } + if (openFile(next)) { + return true; } + } + LOG(error) << "GeneratorFromO2Kine: no further usable input file"; + return false; +} + +GeneratorFromO2Kine::GeneratorFromO2Kine(std::vector const& filenames) +{ + // this generator should leave all dimensions the same as in the incoming kinematics file + setMomentumUnit(1.); + setEnergyUnit(1.); + setPositionUnit(1.); + setTimeUnit(1.); + + mFileNames = filenames; + if (mFileNames.empty()) { + LOG(error) << "GeneratorFromO2Kine: no input file given"; return; } - LOG(error) << "Problem reading events from file " << name; + LOG(info) << "GeneratorFromO2Kine will read from " << mFileNames.size() + << " file(s), one after the other"; + connectToAlienIfNeeded(mFileNames); +} + +GeneratorFromO2Kine::GeneratorFromO2Kine(const char* name) : GeneratorFromO2Kine(splitFileNames(name ? name : "")) +{ +} + +GeneratorFromO2Kine::GeneratorFromO2Kine(O2KineGenConfig const& pars) : GeneratorFromO2Kine(splitFileNames(pars.fileName)) +{ + mConfig = std::make_unique(pars); } -GeneratorFromO2Kine::GeneratorFromO2Kine(O2KineGenConfig const& pars) : GeneratorFromO2Kine(pars.fileName.c_str()) +GeneratorFromO2Kine::GeneratorFromO2Kine(O2KineGenConfig const& pars, std::vector const& filenames) : GeneratorFromO2Kine(filenames) { mConfig = std::make_unique(pars); } +GeneratorFromO2Kine::~GeneratorFromO2Kine() +{ + closeCurrentFile(); +} + bool GeneratorFromO2Kine::Init() { // read and set params LOG(info) << "Init \'FromO2Kine\' generator"; - mSkipNonTrackable = mConfig->skipNonTrackable; - mContinueMode = mConfig->continueMode; - mRoundRobin = mConfig->roundRobin; - mRandomize = mConfig->randomize; - mRngSeed = mConfig->rngseed; - mRandomPhi = mConfig->randomphi; - if (mRandomize) { + if (mConfig) { + mSkipNonTrackable = mConfig->skipNonTrackable; + mContinueMode = mConfig->continueMode; + mRoundRobin = mConfig->roundRobin; + mRandomize = mConfig->randomize; + mRngSeed = mConfig->rngseed; + mRandomPhi = mConfig->randomphi; + } + if (mRandomize && mRngSeed > 0) { + // with a zero the seed given to the driver (o2-sim / o2-sim-dpl-eventgen --seed) stays in control gRandom->SetSeed(mRngSeed); } + mCurrentFileIndex = -1; + if (!openNextFile(false)) { + 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 (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 = getExpectedNEvents(); + if (requested > 0 && !mRoundRobin && mEventsAvailable > 0) { + auto estimate = (size_t)mEventsAvailable * mFileNames.size(); + if (estimate < requested) { + LOG(warn) << "This job will request " << requested << " events, but the input (" + << mFileNames.size() << " file(s), " << mEventsAvailable + << " events in the first one) holds only about " << estimate << ". Unless the " + << "remaining files are larger, the job will stop with 'ran out of events' - " + << "provide more files/events or enable roundRobin"; + } + } return true; } void GeneratorFromO2Kine::SetStartEvent(int start) { - if (start < mEventsAvailable) { - mEventCounter = start; - } else { - LOG(error) << "start event bigger than available events\n"; - } + // this refers to the first file and is applied once that file has been opened + mStartEvent = start; } bool GeneratorFromO2Kine::importParticles() @@ -249,10 +426,26 @@ bool GeneratorFromO2Kine::importParticles() // It might need some adjustment to make it work with secondaries or to continue // from a kinematics snapshot - // Randomize the order of events in the input file + // Next file in the list opened when the events of the current one are used up + if (mEventCounter >= mEventsAvailable) { + if (!openNextFile(mRoundRobin)) { + 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)" : "") + << ". Provide more input files/events or allow reusing them via roundRobin"; + return false; + } + } + if (mCurrentFile == nullptr || mEventBranch == nullptr || mEventCounter >= (int)mEventOrder.size()) { + LOG(fatal) << "GeneratorFromO2Kine: no input file available"; + return false; + } + // the entry to be read from the file which is currently open; the order was fixed + // when the file was opened, so every event of it is used exactly once + auto entry = mEventOrder[mEventCounter]; if (mRandomize) { - mEventCounter = gRandom->Integer(mEventsAvailable); - LOG(info) << "GeneratorFromO2Kine - Picking event " << mEventCounter; + LOG(info) << "GeneratorFromO2Kine - Picking event " << entry; } double dPhi = 0.; @@ -262,81 +455,74 @@ bool GeneratorFromO2Kine::importParticles() LOG(info) << "Rotating phi by " << dPhi; } - if (mEventCounter < mEventsAvailable) { - int particlecounter = 0; - - std::vector* tracks = nullptr; - mEventBranch->SetAddress(&tracks); - mEventBranch->GetEntry(mEventCounter); - - if (mMCHeaderBranch) { - o2::dataformats::MCEventHeader* mcheader = nullptr; - mMCHeaderBranch->SetAddress(&mcheader); - mMCHeaderBranch->GetEntry(mEventCounter); - mOrigMCEventHeader.reset(mcheader); - } - - for (auto& t : *tracks) { + int particlecounter = 0; - // in case we do not want to continue, take only primaries - if (!mContinueMode && !t.isPrimary()) { - continue; - } + std::vector* tracks = nullptr; + mEventBranch->SetAddress(&tracks); + mEventBranch->GetEntry(entry); + mLastEntryRead = entry; - auto pdg = t.GetPdgCode(); - auto px = t.Px(); - auto py = t.Py(); - if (mRandomPhi) { - // transformation applied through rotation matrix - auto cos = TMath::Cos(dPhi); - auto sin = TMath::Sin(dPhi); - auto newPx = px * cos - py * sin; - auto newPy = px * sin + py * cos; - px = newPx; - py = newPy; - } - auto pz = t.Pz(); - auto vx = t.Vx(); - auto vy = t.Vy(); - auto vz = t.Vz(); - auto m1 = t.getMotherTrackId(); - auto m2 = t.getSecondMotherTrackId(); - auto d1 = t.getFirstDaughterTrackId(); - auto d2 = t.getLastDaughterTrackId(); - auto e = t.GetEnergy(); - auto vt = t.T() * 1e-9; // MCTrack stores in ns ... generators and engines use seconds - auto weight = t.getWeight(); - auto wanttracking = t.getToBeDone(); - - if (mContinueMode) { // in case we want to continue, do only inhibited tracks - wanttracking &= t.getInhibited(); - } + if (mMCHeaderBranch) { + o2::dataformats::MCEventHeader* mcheader = nullptr; + mMCHeaderBranch->SetAddress(&mcheader); + mMCHeaderBranch->GetEntry(entry); + mOrigMCEventHeader.reset(mcheader); + } - LOG(debug) << "Putting primary " << pdg; + for (auto& t : *tracks) { - mParticles.push_back(TParticle(pdg, t.getStatusCode().fullEncoding, m1, m2, d1, d2, px, py, pz, e, vx, vy, vz, vt)); - mParticles.back().SetUniqueID((unsigned int)t.getProcess()); // we should propagate the process ID - mParticles.back().SetBit(ParticleStatus::kToBeDone, wanttracking); - mParticles.back().SetWeight(weight); + // in case we do not want to continue, take only primaries + if (!mContinueMode && !t.isPrimary()) { + continue; + } - particlecounter++; + auto pdg = t.GetPdgCode(); + auto px = t.Px(); + auto py = t.Py(); + if (mRandomPhi) { + // transformation applied through rotation matrix + auto cos = TMath::Cos(dPhi); + auto sin = TMath::Sin(dPhi); + auto newPx = px * cos - py * sin; + auto newPy = px * sin + py * cos; + px = newPx; + py = newPy; } - mEventCounter++; - if (mRoundRobin) { - LOG(info) << "Resetting event counter to 0; Reusing events from file"; - mEventCounter = mEventCounter % mEventsAvailable; + auto pz = t.Pz(); + auto vx = t.Vx(); + auto vy = t.Vy(); + auto vz = t.Vz(); + auto m1 = t.getMotherTrackId(); + auto m2 = t.getSecondMotherTrackId(); + auto d1 = t.getFirstDaughterTrackId(); + auto d2 = t.getLastDaughterTrackId(); + auto e = t.GetEnergy(); + auto vt = t.T() * 1e-9; // MCTrack stores in ns ... generators and engines use seconds + auto weight = t.getWeight(); + auto wanttracking = t.getToBeDone(); + + if (mContinueMode) { // in case we want to continue, do only inhibited tracks + wanttracking &= t.getInhibited(); } - if (tracks) { - delete tracks; - } + LOG(debug) << "Putting primary " << pdg; - LOG(info) << "Event generator put " << particlecounter << " on stack"; - return true; - } else { - LOG(error) << "GeneratorFromO2Kine: Ran out of events\n"; + mParticles.push_back(TParticle(pdg, t.getStatusCode().fullEncoding, m1, m2, d1, d2, px, py, pz, e, vx, vy, vz, vt)); + mParticles.back().SetUniqueID((unsigned int)t.getProcess()); // we should propagate the process ID + mParticles.back().SetBit(ParticleStatus::kToBeDone, wanttracking); + mParticles.back().SetWeight(weight); + + particlecounter++; } - return false; + mEventCounter++; + mEventsServed++; + + if (tracks) { + delete tracks; + } + + LOG(info) << "Event generator put " << particlecounter << " on stack"; + return true; } void GeneratorFromO2Kine::updateHeader(o2::dataformats::MCEventHeader* eventHeader) @@ -346,14 +532,14 @@ void GeneratorFromO2Kine::updateHeader(o2::dataformats::MCEventHeader* eventHead // we forward the original header information if any if (mOrigMCEventHeader.get()) { eventHeader->copyInfoFrom(*mOrigMCEventHeader.get()); + // we forward also the original basic vertex information contained in FairMCEventHeader + static_cast(*eventHeader) = static_cast(*mOrigMCEventHeader.get()); } - // we forward also the original basic vertex information contained in FairMCEventHeader - static_cast(*eventHeader) = static_cast(*mOrigMCEventHeader.get()); // put additional information about input file and event number of the current event eventHeader->putInfo("forwarding-generator", "generatorFromO2Kine"); - eventHeader->putInfo("forwarding-generator_inputFile", mEventFile->GetName()); - eventHeader->putInfo("forwarding-generator_inputEventNumber", mEventCounter - 1); + eventHeader->putInfo("forwarding-generator_inputFile", getCurrentFileName()); + eventHeader->putInfo("forwarding-generator_inputEventNumber", mLastEntryRead); } namespace @@ -391,42 +577,54 @@ bool GeneratorFromEventPool::Init() setPositionUnit(1.); setEnergyUnit(1.); - // initialize the event pool + // initialize the event pool. + // When zero is provided as seed, the global ROOT random sequence is followed + // so that the seed given to the o2-sim or o2-sim-dpl-eventgen + // also determines which files of the pool the job will pick if (mConfig.rngseed > 0) { mRandomEngine.seed(mConfig.rngseed); } else { - std::random_device rd; - mRandomEngine.seed(rd()); + mRandomEngine.seed(gRandom->Integer(std::numeric_limits::max())); } TString expPath(mConfig.eventPoolPath); gSystem->ExpandPathName(expPath); 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"; - // now choose the actual file - std::uniform_int_distribution distribution(0, mPoolFilesAvailable.size() - 1); - auto chosenIndex = distribution(mRandomEngine); - mFileChosen = mPoolFilesAvailable[chosenIndex]; - LOG(info) << "EventPool is using file " << mFileChosen; + // shuffle the pool so that different jobs go through it in a different order + mFilesChosen = selectFiles(mPoolFilesAvailable); + LOG(info) << "EventPool will go through all " << mFilesChosen.size() << " pool files"; - // we bring up the internal mO2KineGenerator + // we bring up the internal mO2KineGenerator with the shuffled file list auto kine_config = O2KineGenConfig{ .skipNonTrackable = mConfig.skipNonTrackable, .continueMode = false, - .roundRobin = false, + .roundRobin = mConfig.roundRobin, .randomize = mConfig.randomize, .rngseed = mConfig.rngseed, - .randomphi = mConfig.randomphi, - .fileName = mFileChosen}; - mO2KineGenerator.reset(new GeneratorFromO2Kine(kine_config)); + .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 GeneratorFromEventPool::selectFiles(std::vector const& universe) +{ + // shuffles the whole pool universe so that different jobs go through it in a + // 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; +} + namespace { namespace fs = std::filesystem; diff --git a/Generators/src/GeneratorHepMC.cxx b/Generators/src/GeneratorHepMC.cxx index 10646d4a8379d..113cf9185a2a2 100644 --- a/Generators/src/GeneratorHepMC.cxx +++ b/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 " diff --git a/Generators/src/GeneratorHybrid.cxx b/Generators/src/GeneratorHybrid.cxx index ed831aef8b16a..b77a27faaf25c 100644 --- a/Generators/src/GeneratorHybrid.cxx +++ b/Generators/src/GeneratorHybrid.cxx @@ -18,6 +18,7 @@ #include #include #include "TGrid.h" +#include "TROOT.h" namespace o2 { @@ -84,11 +85,13 @@ GeneratorHybrid::GeneratorHybrid(const std::string& inputgens) return; } } - // Check if all elements of mFractions are 0 - if (std::all_of(mFractions.begin(), mFractions.end(), [](int i) { return i == 0; })) { - LOG(fatal) << "All fractions provided are 0, no simulation will be performed"; - return; - } + } + // Check if all elements of mFractions are 0 (currently not in parallel mode, which does not use the fractions) + // To-do: use the fractions also in parallel mode for sub-generators of different kinds + if (mGenerationMode != GenMode::kParallel && + std::all_of(mFractions.begin(), mFractions.end(), [](int i) { return i == 0; })) { + LOG(fatal) << "All fractions provided are 0, no simulation will be performed"; + return; } for (auto gen : mInputGens) { // Search if the generator name is inside generatorNames (which is a vector of strings) @@ -187,8 +190,82 @@ GeneratorHybrid::~GeneratorHybrid() mStopFlag = true; } +std::vector GeneratorHybrid::expectedNEventsPerGenerator(unsigned int nevents) const +{ + std::vector expected(gens.size(), 0); + // in parallel mode any sub-generator can serve any event: the share is not known + if (nevents == 0 || mGenerationMode == GenMode::kParallel) { + return expected; + } + // a fraction is given per generator, or per group in cocktail mode + const size_t nslots = mCocktailMode ? mGroups.size() : gens.size(); + if (mFractions.size() != nslots) { + return expected; + } + unsigned int sum = 0; + for (auto f : mFractions) { + sum += std::max(f, 0); + } + if (sum == 0) { + return expected; + } + std::vector perSlot(nslots, 0); + if (mRandomize) { + // the slot is drawn for each event with probability fraction/sum: use the mean, rounded up + for (size_t k = 0; k < nslots; ++k) { + const unsigned long long f = std::max(mFractions[k], 0); + perSlot[k] = (unsigned int)((f * nevents + sum - 1) / sum); + } + } else { + // the slots are used in turn, mFractions[k] consecutive events each (see generateEvent) + const unsigned int cycles = nevents / sum; + unsigned int rest = nevents % sum; + for (size_t k = 0; k < nslots; ++k) { + const unsigned int f = std::max(mFractions[k], 0); + const unsigned int inLastCycle = std::min(f, rest); + perSlot[k] = cycles * f + inLastCycle; + rest -= inLastCycle; + } + } + for (size_t k = 0; k < nslots; ++k) { + if (mCocktailMode) { + // every member of a group contributes to each event of the group + for (auto subIndex : mGroups[k]) { + expected[subIndex] = perSlot[k]; + } + } else { + expected[k] = perSlot[k]; + } + } + return expected; +} + +void GeneratorHybrid::requestEvent(int genIndex) +{ + if (!mRequestBudget.empty()) { + if (mRequestBudget[genIndex] == 0) { + // the sub-generator was already asked for its whole share of the job + return; + } + mRequestBudget[genIndex]--; + } + mInputTaskQueue.push(genIndex); + mTasksStarted++; +} + +void GeneratorHybrid::rethrowSubGeneratorFailure(int genIndex) +{ + if (mGenFailures[genIndex]) { + LOG(error) << "Sub-generator " << genIndex << " (" << mGens[genIndex] << ") failed to generate the requested event"; + mStopFlag = true; + std::rethrow_exception(mGenFailures[genIndex]); + } +} + Bool_t GeneratorHybrid::Init() { + // number of events each sub-generator will serve, derived from the fractions + const auto expectedNEvents = expectedNEventsPerGenerator(getExpectedNEvents()); // init all sub-gens int count = 0; for (auto& gen : mGens) { @@ -209,7 +286,12 @@ Bool_t GeneratorHybrid::Init() LOG(info) << "Setting \'Pythia8\' base configuration: " << config << std::endl; dynamic_cast(gens[count].get())->setConfig(config); } - gens[count]->Init(); // TODO: move this to multi-threaded + // inside the hybrid a sub-generator only serves its share of the job's events + gens[count]->setExpectedNEvents(expectedNEvents[count]); + if (!gens[count]->Init()) { // TODO: move this to multi-threaded + LOG(fatal) << "Initialization of sub-generator " << count << " (" << gen << ") failed"; + return kFALSE; + } addSubGenerator(count, gen); if (mTriggerModes[count] != o2::eventgen::Generator::kTriggerOFF) { gens[count]->setTriggerMode(mTriggerModes[count]); @@ -285,8 +367,15 @@ Bool_t GeneratorHybrid::Init() } else { LOG(info) << "Generators will be used in sequence, following provided fractions"; } + // With fixed fractions the share of each sub-generator is exact: a sub-generator is never asked for more + // events than that. Otherwise, an event pool holding exactly its share and used up before the end of the job + // would be asked for an additional event. + if (!mRandomize && mGenerationMode != GenMode::kParallel && getExpectedNEvents() > 0) { + mRequestBudget = expectedNEvents; + } mGenIsInitialized.resize(gens.size(), false); + mGenFailures.resize(gens.size()); if (mGenerationMode == GenMode::kParallel) { // in parallel mode we just use one queue --> collaboration mResultQueue.resize(1); @@ -294,6 +383,9 @@ Bool_t GeneratorHybrid::Init() // in sequential mode we have one queue per generator mResultQueue.resize(gens.size()); } + // The sub-generators run in worker threads and may use ROOT concurrently such as in two event pools reading. + // This is done in o2-sim but not in o2-sim-dpl-eventgen. Done here as Hybrid creates the worker threads + ROOT::EnableThreadSafety(); // Create a task arena with a specified number of threads mTaskArena.initialize(GeneratorHybridParam::Instance().num_workers); @@ -312,12 +404,18 @@ Bool_t GeneratorHybrid::Init() // mGenIsInitialized[task] = true; // } } - bool isTriggered = false; - while (!isTriggered) { - generator->clearParticles(); - generator->generateEvent(); - generator->importParticles(); - isTriggered = generator->triggerEvent(); + // A failure must not escape the worker: the other workers would keep polling and the job would hang. + // It is handed over to the main thread with the result, and raised there if the event is consumed. + try { + bool isTriggered = false; + while (!isTriggered) { + generator->clearParticles(); + generator->generateEvent(); + generator->importParticles(); + isTriggered = generator->triggerEvent(); + } + } catch (...) { + mGenFailures[task] = std::current_exception(); } LOG(debug) << "eventgen finished for task " << task; if (!mStopFlag) { @@ -353,8 +451,7 @@ Bool_t GeneratorHybrid::Init() // let's also push initial generation tasks for each event generator for (size_t genindex = 0; genindex < gens.size(); ++genindex) { - mInputTaskQueue.push(genindex); - mTasksStarted++; + requestEvent(genindex); } mIsInitialized = true; return Generator::Init(); @@ -422,6 +519,13 @@ bool GeneratorHybrid::importParticles() } } } + if (mCocktailMode && mIndex != -1) { + for (auto subIndex : subGenIndex) { + rethrowSubGeneratorFailure(subIndex); + } + } else { + rethrowSubGeneratorFailure(genIndex); + } auto unit_transformer = [](auto& p, auto pos_unit, auto time_unit, auto en_unit, auto mom_unit) { p.SetMomentum(p.Px() * mom_unit, p.Py() * mom_unit, p.Pz() * mom_unit, p.Energy() * en_unit); @@ -443,6 +547,10 @@ bool GeneratorHybrid::importParticles() } }; + // The event consumed here is the last one the job asks for. It is important not to schedule a lookahead + // event for the sub-generators, which could never be consumed (or not exist as in the case of limited pools). + const bool lastEvent = getTotalNEvents() > 0 && (mEventCounter + 1) >= static_cast(getTotalNEvents()); + // Clear particles and event header mParticles.clear(); // event header of underlying generator must be fully reset @@ -477,8 +585,9 @@ bool GeneratorHybrid::importParticles() gens[subIndex]->updateHeader(&mMCEventHeader); mHeaderGeneratorIndex = subIndex; // store index of generator updating the header } - mInputTaskQueue.push(subIndex); - mTasksStarted++; + if (!lastEvent) { + requestEvent(subIndex); + } } } else { LOG(info) << "Importing particles for task " << genIndex; @@ -499,8 +608,9 @@ bool GeneratorHybrid::importParticles() // fetch the event Header information from the underlying generator gens[genIndex]->updateHeader(&mMCEventHeader); mHeaderGeneratorIndex = genIndex; // store index of generator updating the header - mInputTaskQueue.push(genIndex); - mTasksStarted++; + if (!lastEvent) { + requestEvent(genIndex); + } } mseqCounter++; @@ -719,8 +829,8 @@ Bool_t GeneratorHybrid::parseJSON(const std::string& path) if (doc.HasMember("fractions")) { const auto& fractions = doc["fractions"]; for (const auto& frac : fractions.GetArray()) { - if (!frac.IsInt()) { - LOG(fatal) << "Fractions must be integers. Wrong type found in JSON"; + if (!frac.IsInt() || frac.GetInt() < 0) { + LOG(fatal) << "Fractions must be non-negative integers. Wrong value found in JSON"; return false; } mFractions.push_back(frac.GetInt()); diff --git a/Generators/test/EventPoolTestUtils.h b/Generators/test/EventPoolTestUtils.h new file mode 100644 index 0000000000000..1bc9ae9a5afbf --- /dev/null +++ b/Generators/test/EventPoolTestUtils.h @@ -0,0 +1,187 @@ +// Copyright 2019-2026 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +/// \file EventPoolTestUtils.h +/// \brief helpers shared by the unit tests of the event pool chaining / hybrid generator +/// \author M. Giacalone, mgiacalo@cern.ch, 08/2026 + +#ifndef ALICEO2_EVENTGEN_TEST_EVENTPOOLTESTUTILS_H_ +#define ALICEO2_EVENTGEN_TEST_EVENTPOOLTESTUTILS_H_ + +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace evtpooltest +{ +namespace fs = std::filesystem; + +/// the px of the first track of an event encodes the file it belongs to and its +/// position within that file; this allows to check the reading order later on +inline double encodeMomentum(int fileTag, int event) +{ + return 1000. * fileTag + event; +} + +/// creates a minimal - but structurally valid - O2 kinematics file (optionally without +/// the MCEventHeader branch); +/// event `ev` contains `ev + 1` primary tracks (optionally linked as a chain) +inline void createKineFile(std::string const& path, int nevents, int fileTag, bool withHeader = true, bool chain = false) +{ + std::unique_ptr file(TFile::Open(path.c_str(), "RECREATE")); + auto tree = new TTree("o2sim", "o2sim"); // owned by the file + std::vector tracks; + tree->Branch("MCTrack", &tracks); + o2::dataformats::MCEventHeader header; + auto headerPtr = &header; + if (withHeader) { + tree->Branch("MCEventHeader.", &headerPtr); + } + + for (int ev = 0; ev < nevents; ++ev) { + tracks.clear(); + for (int i = 0; i <= ev; ++i) { + // with `chain` the tracks of an event form a chain (track i is the mother of track i + 1), which + // allows to check that the mother/daughter indices survive the merging done by a cocktail + int mother = (chain && i > 0) ? i - 1 : -1; + int daughter = (chain && i < ev) ? i + 1 : -1; + o2::MCTrack track(211, mother, -1, daughter, daughter, encodeMomentum(fileTag, ev), 0., 0., 0., 0., 0., 0., 0); + track.setToBeDone(true); + tracks.push_back(track); + } + header.Reset(); + header.SetEventID(static_cast(encodeMomentum(fileTag, ev))); + header.SetVertex(0., 0., 0.); + header.putInfo("test_fileTag", fileTag); + tree->Fill(); + } + tree->Write(); + file->Close(); +} + +/// creates `nfiles` event pool files under `//evtpool.root`; +/// file i holds i + 2 events +inline std::vector createPool(fs::path const& tmpDir, int nfiles, bool chain = false) +{ + std::vector filenames; + for (int i = 0; i < nfiles; ++i) { + auto fileDir = tmpDir / std::to_string(i); + fs::create_directories(fileDir); + auto filePath = fileDir / o2::eventgen::GeneratorFromEventPool::eventpool_filename; + createKineFile(filePath.string(), i + 2, i, /*withHeader=*/true, chain); + filenames.push_back(filePath.string()); + } + return filenames; +} + +/// number of events in pool file i (must match createPool) +inline int eventsInFile(int i) { return i + 2; } + +/// scratch directory that removes itself again; it has to be declared before the +/// generators using it, so that the generators (and with them the open files) are +/// destructed first +struct TempDir { + explicit TempDir(std::string const& tag) + { + path = fs::temp_directory_path() / (tag + "_" + std::to_string(getpid()) + "_" + std::to_string(std::rand())); + std::error_code ec; + fs::remove_all(path, ec); + fs::create_directories(path); + } + ~TempDir() + { + std::error_code ec; + fs::remove_all(path, ec); + } + fs::path path; +}; + +/// number of ROOT files currently open in the process +inline int openRootFiles() +{ + return gROOT->GetListOfFiles() ? gROOT->GetListOfFiles()->GetEntries() : 0; +} + +/// reads the next event and returns the identifier encoded in the first track +inline double readNextEvent(o2::eventgen::Generator& gen) +{ + gen.clearParticles(); + if (!gen.importParticles()) { + return -1.; + } + auto const& particles = gen.getParticles(); + if (particles.empty()) { + return -1.; + } + return particles.front().Px(); +} + +/// Records the log messages of at least the given severity while it is alive, and makes LOG(fatal) +/// throw a fair::FatalException - which is what the O2 device runners do - instead of taking a +/// core dump. Without such a handler a fatal condition cannot be asserted from within a unit test. +/// The sink is called from whatever thread is logging, hence the mutex. +class LogCatcher +{ + public: + explicit LogCatcher(std::string const& key, fair::Severity severity = fair::Severity::warn) : mKey(key) + { + fair::Logger::OnFatal([] { throw fair::FatalException("fatal condition (LOG(fatal)) raised in a unit test"); }); + fair::Logger::AddCustomSink(mKey, severity, [this](std::string const& content, fair::LogMetaData const&) { + std::lock_guard lock(mMutex); + mMessages.push_back(content); + }); + } + ~LogCatcher() { fair::Logger::RemoveCustomSink(mKey); } + LogCatcher(LogCatcher const&) = delete; + LogCatcher& operator=(LogCatcher const&) = delete; + + /// number of recorded messages containing the given text + int count(std::string const& text) const + { + std::lock_guard lock(mMutex); + return std::count_if(mMessages.begin(), mMessages.end(), [&](std::string const& m) { return m.find(text) != std::string::npos; }); + } + /// the first recorded message containing the given text ("" if none) + std::string find(std::string const& text) const + { + std::lock_guard lock(mMutex); + for (auto const& m : mMessages) { + if (m.find(text) != std::string::npos) { + return m; + } + } + return ""; + } + + private: + std::string mKey; + mutable std::mutex mMutex; + std::vector mMessages; +}; + +} // namespace evtpooltest + +#endif diff --git a/Generators/test/test_EventPoolChaining.cxx b/Generators/test/test_EventPoolChaining.cxx new file mode 100644 index 0000000000000..d7df408eb3678 --- /dev/null +++ b/Generators/test/test_EventPoolChaining.cxx @@ -0,0 +1,952 @@ +// Copyright 2019-2026 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +/// \file test_EventPoolChaining.cxx +/// \brief tests reading several (event pool) kinematics files one after the other +/// in GeneratorFromO2Kine and GeneratorFromEventPool +/// \author M. Giacalone, mgiacalo@cern.ch, 08/2026 + +#define BOOST_TEST_MODULE Test EventPoolChaining +#define BOOST_TEST_MAIN +#define BOOST_TEST_DYN_LINK +#include + +#include + +#include "EventPoolTestUtils.h" +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace fs = std::filesystem; + +using namespace evtpooltest; + +/// several files are read one after the other, in the order in which they were given +BOOST_AUTO_TEST_CASE(Rollover_MultipleFiles_Sequential) +{ + TempDir tmpDirGuard("rollover_sequential"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 3; + auto filenames = createPool(tmpDir, numfiles); + + o2::eventgen::GeneratorFromO2Kine gen(filenames); + BOOST_CHECK_EQUAL(gen.getNumberOfFiles(), numfiles); + BOOST_CHECK(gen.Init()); + // only the first file is known/open at this point + BOOST_CHECK_EQUAL(gen.getEventsAvailable(), eventsInFile(0)); + BOOST_CHECK_EQUAL(gen.getCurrentFileIndex(), 0); + + // events must come out file by file, in order + for (int file = 0; file < numfiles; ++file) { + for (int ev = 0; ev < eventsInFile(file); ++ev) { + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(file, ev), 1E-6); + BOOST_CHECK_EQUAL(gen.getCurrentFileIndex(), file); + } + } + // all files are used up now; asking for more is a fatal condition, which cannot be + // checked from within the process (see the example script for the end-to-end check) + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), numfiles); +} + +/// the files must be opened lazily: only when the events of the current one are +/// exhausted, and never more than one at a time +BOOST_AUTO_TEST_CASE(Rollover_OpensFilesLazily) +{ + TempDir tmpDirGuard("rollover_lazy"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 4; + auto filenames = createPool(tmpDir, numfiles); + + auto filesOpenBefore = openRootFiles(); + + o2::eventgen::GeneratorFromO2Kine gen(filenames); + // the constructor must not open anything at all + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 0); + BOOST_CHECK(gen.Init()); + // exactly one input file is open, no matter how many were given + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), 1); + + // reading the events of the first file must not touch any other file + for (int ev = 0; ev < eventsInFile(0); ++ev) { + readNextEvent(gen); + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), 1); + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + } + + // the next event triggers opening the second file - and only the second one + readNextEvent(gen); + BOOST_CHECK_EQUAL(gen.getCurrentFileIndex(), 1); + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), 2); + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); +} + +/// a file which cannot be read is only discovered - and skipped - when it is reached +BOOST_AUTO_TEST_CASE(Rollover_SkipsBadFilesLazily) +{ + TempDir tmpDirGuard("rollover_badfiles"); + auto const& tmpDir = tmpDirGuard.path; + auto good0 = (tmpDir / "good0.root").string(); + auto good1 = (tmpDir / "good1.root").string(); + createKineFile(good0, 2, 0); + createKineFile(good1, 2, 1); + auto nonexisting = (tmpDir / "doesnotexist.root").string(); + + // the broken file sits in the middle of the list: construction must succeed + o2::eventgen::GeneratorFromO2Kine gen({good0, nonexisting, good1}); + BOOST_CHECK_EQUAL(gen.getNumberOfFiles(), 3); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_EQUAL(gen.getCurrentFileIndex(), 0); + + // the two events of the first file, then the broken one is skipped + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(0, 0), 1E-6); + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(0, 1), 1E-6); + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(1, 0), 1E-6); + BOOST_CHECK_EQUAL(gen.getCurrentFileIndex(), 2); + + // nothing usable at all -> Init must stop the job (LOG(fatal)), rather than fail silently and crash later: + // FairPrimaryGenerator ignores the result of Init() + LogCatcher log("badfiles_init", fair::Severity::fatal); + o2::eventgen::GeneratorFromO2Kine badgen({nonexisting}); + BOOST_CHECK_THROW(badgen.Init(), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("none of the 1 input file(s) can be used"), 1); +} + +/// the number of particles per event must be preserved across the file boundaries +BOOST_AUTO_TEST_CASE(Rollover_ParticleContent) +{ + TempDir tmpDirGuard("rollover_content"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 3; + auto filenames = createPool(tmpDir, numfiles); + + o2::eventgen::GeneratorFromO2Kine gen(filenames); + BOOST_CHECK(gen.Init()); + + for (int file = 0; file < numfiles; ++file) { + for (int ev = 0; ev < eventsInFile(file); ++ev) { + gen.clearParticles(); + BOOST_CHECK(gen.importParticles()); + BOOST_CHECK_EQUAL(gen.getParticles().size(), static_cast(ev + 1)); + } + } +} + +/// the MC event header of the original file must be forwarded also when reading +/// from a file that is not the first one of the list +BOOST_AUTO_TEST_CASE(Rollover_HeaderForwarding) +{ + TempDir tmpDirGuard("rollover_header"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 3; + auto filenames = createPool(tmpDir, numfiles); + + o2::eventgen::GeneratorFromO2Kine gen(filenames); + BOOST_CHECK(gen.Init()); + + for (int file = 0; file < numfiles; ++file) { + for (int ev = 0; ev < eventsInFile(file); ++ev) { + gen.clearParticles(); + BOOST_CHECK(gen.importParticles()); + + o2::dataformats::MCEventHeader header; + gen.updateHeader(&header); + BOOST_CHECK_EQUAL(static_cast(header.GetEventID()), static_cast(encodeMomentum(file, ev))); + + bool isvalid = false; + auto tag = header.getInfo("test_fileTag", isvalid); + BOOST_CHECK(isvalid); + BOOST_CHECK_EQUAL(tag, file); + + // the bookkeeping information must point to the file the event was read from + auto inputFile = header.getInfo("forwarding-generator_inputFile", isvalid); + BOOST_CHECK(isvalid); + BOOST_CHECK_EQUAL(inputFile, filenames[file]); + + // ... and to the entry within that very file + auto entry = header.getInfo("forwarding-generator_inputEventNumber", isvalid); + BOOST_CHECK(isvalid); + BOOST_CHECK_EQUAL(entry, ev); + } + } +} + +/// a comma-separated list of file names is read one file after the other as well +BOOST_AUTO_TEST_CASE(Rollover_CommaSeparatedFileNames) +{ + TempDir tmpDirGuard("rollover_commalist"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 3; + auto filenames = createPool(tmpDir, numfiles); + + std::string joined; + for (auto const& f : filenames) { + joined += (joined.empty() ? "" : ",") + f; + } + + auto splitted = o2::eventgen::GeneratorFromO2Kine::splitFileNames(joined); + BOOST_CHECK_EQUAL(splitted.size(), static_cast(numfiles)); + // white space around the separators must be tolerated + BOOST_CHECK_EQUAL(o2::eventgen::GeneratorFromO2Kine::splitFileNames(" a.root , b.root ,,").size(), 2u); + + o2::eventgen::GeneratorFromO2Kine gen(joined.c_str()); + BOOST_CHECK_EQUAL(gen.getNumberOfFiles(), numfiles); + BOOST_CHECK(gen.Init()); + for (int file = 0; file < numfiles; ++file) { + for (int ev = 0; ev < eventsInFile(file); ++ev) { + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(file, ev), 1E-6); + } + } +} + +/// round robin must wrap around the whole file list, not around a single file +BOOST_AUTO_TEST_CASE(Rollover_RoundRobin) +{ + TempDir tmpDirGuard("rollover_roundrobin"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 2; + auto filenames = createPool(tmpDir, numfiles); + const int total = eventsInFile(0) + eventsInFile(1); + + o2::eventgen::O2KineGenConfig config; + config.roundRobin = true; + o2::eventgen::GeneratorFromO2Kine gen(config, filenames); + BOOST_CHECK(gen.Init()); + + // read twice as many events as available; the second pass must repeat the first one + std::vector firstPass; + for (int i = 0; i < total; ++i) { + firstPass.push_back(readNextEvent(gen)); + } + for (int i = 0; i < total; ++i) { + BOOST_CHECK_CLOSE(readNextEvent(gen), firstPass[i], 1E-6); + } +} + +/// the event header of one file must not be forwarded for the events of the next file, +/// in case that one carries no MCEventHeader at all +BOOST_AUTO_TEST_CASE(Rollover_NoStaleHeaderAcrossFiles) +{ + TempDir tmpDirGuard("rollover_staleheader"); + auto const& tmpDir = tmpDirGuard.path; + auto withHeader = (tmpDir / "withheader.root").string(); + auto noHeader = (tmpDir / "noheader.root").string(); + createKineFile(withHeader, 2, 7); + createKineFile(noHeader, 2, 8, /*withHeader=*/false); + + o2::eventgen::GeneratorFromO2Kine gen({withHeader, noHeader}); + BOOST_CHECK(gen.Init()); + + bool isvalid = false; + for (int ev = 0; ev < 2; ++ev) { + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(7, ev), 1E-6); + o2::dataformats::MCEventHeader header; + gen.updateHeader(&header); + BOOST_CHECK_EQUAL(header.getInfo("test_fileTag", isvalid), 7); + BOOST_CHECK(isvalid); + } + for (int ev = 0; ev < 2; ++ev) { + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(8, ev), 1E-6); + o2::dataformats::MCEventHeader header; + gen.updateHeader(&header); + // nothing of the first file may show up here + header.getInfo("test_fileTag", isvalid); + BOOST_CHECK(!isvalid); + BOOST_CHECK(static_cast(header.GetEventID()) != static_cast(encodeMomentum(7, 0))); + BOOST_CHECK(static_cast(header.GetEventID()) != static_cast(encodeMomentum(7, 1))); + // the bookkeeping still refers to the file actually read + BOOST_CHECK_EQUAL(header.getInfo("forwarding-generator_inputFile", isvalid), noHeader); + BOOST_CHECK_EQUAL(header.getInfo("forwarding-generator_inputEventNumber", isvalid), ev); + } +} + +/// a file which fails to open must not destroy the file that is currently open: +/// with roundRobin the open file is restarted in place, without being reopened +BOOST_AUTO_TEST_CASE(Rollover_FailedOpenKeepsCurrentFile) +{ + TempDir tmpDirGuard("rollover_keepfile"); + auto const& tmpDir = tmpDirGuard.path; + auto good = (tmpDir / "good.root").string(); + createKineFile(good, 3, 0); + auto nonexisting = (tmpDir / "doesnotexist.root").string(); + + o2::eventgen::O2KineGenConfig config; + config.roundRobin = true; + o2::eventgen::GeneratorFromO2Kine gen(config, {good, nonexisting}); + auto filesOpenBefore = openRootFiles(); + BOOST_CHECK(gen.Init()); + + // three passes over the only usable file + for (int pass = 0; pass < 3; ++pass) { + for (int ev = 0; ev < 3; ++ev) { + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(0, ev), 1E-6); + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + } + } + // the good file has been opened exactly once, it was never closed and reopened + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), 1); + BOOST_CHECK_EQUAL(gen.getCurrentFileIndex(), 0); +} + +/// the wrap-around to the beginning of the list is reported also when it happens +/// after skipping an unusable file at the end of the list +BOOST_AUTO_TEST_CASE(Rollover_WrapReportedAfterSkippedFile) +{ + TempDir tmpDirGuard("rollover_wrapmessage"); + auto const& tmpDir = tmpDirGuard.path; + auto good0 = (tmpDir / "good0.root").string(); + auto good1 = (tmpDir / "good1.root").string(); + createKineFile(good0, 2, 0); + createKineFile(good1, 2, 1); + auto nonexisting = (tmpDir / "doesnotexist.root").string(); + + o2::eventgen::O2KineGenConfig config; + config.roundRobin = true; + o2::eventgen::GeneratorFromO2Kine gen(config, {good0, good1, nonexisting}); + LogCatcher log("rollover_wrapmessage", fair::Severity::info); + BOOST_CHECK(gen.Init()); + for (int i = 0; i < 4; ++i) { + readNextEvent(gen); + } + BOOST_CHECK_EQUAL(log.count("Reached the end of the input file list"), 0); + // the last file is skipped, then the search wraps to the first one + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(0, 0), 1E-6); + BOOST_CHECK_EQUAL(log.count("Reached the end of the input file list"), 1); +} + +/// the start event refers to an entry of the first file: it is applied in sequential +/// order only, with randomize it is ignored (with a warning) and every event is served +BOOST_AUTO_TEST_CASE(StartEvent_OnlyWithoutRandomize) +{ + TempDir tmpDirGuard("start_event"); + auto file = (tmpDirGuard.path / "kine.root").string(); + constexpr int nevents = 5; + createKineFile(file, nevents, 0); + { + o2::eventgen::GeneratorFromO2Kine gen(std::vector{file}); + gen.SetStartEvent(2); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(0, 2), 1E-6); + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(0, 3), 1E-6); + } + { + LogCatcher log("start_event_random", fair::Severity::warn); + o2::eventgen::O2KineGenConfig config; + config.randomize = true; + o2::eventgen::GeneratorFromO2Kine gen(config, {file}); + gen.SetStartEvent(2); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_EQUAL(log.count("Start event 2 ignored"), 1); + std::set seen; + for (int ev = 0; ev < nevents; ++ev) { + seen.insert(readNextEvent(gen)); + } + BOOST_CHECK_EQUAL(seen.size(), static_cast(nevents)); + } +} + +/// the number of events a generator expects to serve defaults to the one of the job, +/// but can be set (or declared unknown with 0), as done for sub-generators of a hybrid +BOOST_AUTO_TEST_CASE(ExpectedNEvents) +{ + unsigned int total = 42; + o2::eventgen::Generator::setTotalNEvents(total); + TempDir tmpDirGuard("expected_nevents"); + auto filenames = createPool(tmpDirGuard.path, 1); + + o2::eventgen::GeneratorFromO2Kine gen(filenames); + BOOST_CHECK_EQUAL(gen.getExpectedNEvents(), 42u); + gen.setExpectedNEvents(0); + BOOST_CHECK_EQUAL(gen.getExpectedNEvents(), 0u); + gen.setExpectedNEvents(7); + BOOST_CHECK_EQUAL(gen.getExpectedNEvents(), 7u); + + // the internal generator of the event pool follows the one of the pool + o2::eventgen::EventPoolGenConfig poolConfig; + poolConfig.eventPoolPath = tmpDirGuard.path.string(); + o2::eventgen::GeneratorFromEventPool pool(poolConfig); + pool.setExpectedNEvents(0); + BOOST_CHECK(pool.Init()); + BOOST_CHECK_EQUAL(pool.getO2KineGenerator()->getExpectedNEvents(), 0u); + + total = 0; + o2::eventgen::Generator::setTotalNEvents(total); +} + +/// a single file, read without round robin, must NOT be silently reopened/reused once +/// exhausted: openNextFile()'s "next == mCurrentFileIndex && mCurrentFile" shortcut +/// (which restarts the currently open file in place) may only ever fire when wrapAround +/// (i.e. roundRobin) is true; with roundRobin off, every event is served exactly once and +/// asking for one more is a fatal condition +BOOST_AUTO_TEST_CASE(SingleFile_NoRoundRobin_ExhaustionIsFatal) +{ + LogCatcher log("exhaustion_single", fair::Severity::fatal); + TempDir tmpDirGuard("single_file_no_rr"); + auto const& tmpDir = tmpDirGuard.path; + auto file = (tmpDir / "kine.root").string(); + constexpr int nevents = 3; + createKineFile(file, nevents, 0); + + o2::eventgen::O2KineGenConfig config; + config.roundRobin = false; + o2::eventgen::GeneratorFromO2Kine gen(config, {file}); + BOOST_CHECK(gen.Init()); + + // all events of the single file are served normally, without any repeats + std::set seen; + for (int i = 0; i < nevents; ++i) { + auto id = readNextEvent(gen); + BOOST_CHECK(id >= 0.); + BOOST_CHECK(seen.insert(id).second); + } + BOOST_CHECK_EQUAL(seen.size(), static_cast(nevents)); + BOOST_CHECK_EQUAL(log.count("ran out of events"), 0); + + // the next request is fatal, and so is every further one: the file is not silently restarted + BOOST_CHECK_THROW(readNextEvent(gen), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("ran out of events after 3 event(s) from 1 input file(s)"), 1); + BOOST_CHECK_THROW(readNextEvent(gen), fair::FatalException); + BOOST_CHECK_EQUAL(gen.getEventsServed(), nevents); + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), 1); +} + +/// the pool of several files is fatal only after its LAST file is used up, and the message +/// reports what was served and (if known) what was requested +BOOST_AUTO_TEST_CASE(MultipleFiles_NoRoundRobin_ExhaustionIsFatal) +{ + LogCatcher log("exhaustion_multi", fair::Severity::fatal); + TempDir tmpDirGuard("multi_file_no_rr"); + constexpr int numfiles = 3; + auto filenames = createPool(tmpDirGuard.path, numfiles); + int total = 0; + for (int i = 0; i < numfiles; ++i) { + total += eventsInFile(i); + } + + o2::eventgen::GeneratorFromO2Kine gen(filenames); + gen.setExpectedNEvents(total + 3); + BOOST_CHECK(gen.Init()); + for (int i = 0; i < total; ++i) { + BOOST_CHECK(readNextEvent(gen) >= 0.); + } + BOOST_CHECK_EQUAL(log.count("ran out of events"), 0); + BOOST_CHECK_THROW(readNextEvent(gen), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("ran out of events after " + std::to_string(total) + " event(s) from 3 input file(s) (" + std::to_string(total + 3) + " were requested)"), 1); + BOOST_CHECK_EQUAL(gen.getEventsServed(), total); + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), numfiles); + + // an unknown number of requested events is not quoted + LogCatcher log2("exhaustion_multi_unknown", fair::Severity::fatal); + o2::eventgen::GeneratorFromO2Kine gen2(filenames); + gen2.setExpectedNEvents(0); + BOOST_CHECK(gen2.Init()); + for (int i = 0; i < total; ++i) { + BOOST_CHECK(readNextEvent(gen2) >= 0.); + } + BOOST_CHECK_THROW(readNextEvent(gen2), fair::FatalException); + BOOST_CHECK_EQUAL(log2.count("ran out of events after " + std::to_string(total) + " event(s) from 3 input file(s)."), 1); + BOOST_CHECK_EQUAL(log2.count("were requested"), 0); +} + +/// the same for the event pool generator, which is what -g evtpool / the hybrid generator use +BOOST_AUTO_TEST_CASE(EventPool_NoRoundRobin_ExhaustionIsFatal) +{ + LogCatcher log("exhaustion_pool", fair::Severity::fatal); + TempDir tmpDirGuard("pool_no_rr"); + constexpr int numfiles = 3; + createPool(tmpDirGuard.path, numfiles); + int total = 0; + for (int i = 0; i < numfiles; ++i) { + total += eventsInFile(i); + } + + o2::eventgen::EventPoolGenConfig poolConfig; + poolConfig.eventPoolPath = tmpDirGuard.path.string(); + poolConfig.roundRobin = false; + poolConfig.randomize = true; + poolConfig.rngseed = 17; + o2::eventgen::GeneratorFromEventPool pool(poolConfig); + BOOST_REQUIRE(pool.Init()); + + // every event of the pool is served exactly once ... + std::set seen; + for (int i = 0; i < total; ++i) { + auto id = readNextEvent(pool); + BOOST_CHECK(id >= 0.); + BOOST_CHECK(seen.insert(id).second); + } + BOOST_CHECK_EQUAL(seen.size(), static_cast(total)); + // ... and asking for one more is fatal + BOOST_CHECK_THROW(readNextEvent(pool), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("ran out of events after " + std::to_string(total) + " event(s) from 3 input file(s)"), 1); +} + +/// a pool path which holds no pool file stops the job at initialisation (LOG(fatal)), as a missing input did +/// before the chaining was introduced; the generator would otherwise be used uninitialised and crash +BOOST_AUTO_TEST_CASE(EventPool_NoPoolFile_InitIsFatal) +{ + LogCatcher log("pool_nofile", fair::Severity::fatal); + TempDir tmpDirGuard("pool_nofile"); + o2::eventgen::EventPoolGenConfig poolConfig; + poolConfig.eventPoolPath = (tmpDirGuard.path / "does_not_exist").string(); + o2::eventgen::GeneratorFromEventPool pool(poolConfig); + BOOST_CHECK_THROW(pool.Init(), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("No file found that can be used with EventPool generator"), 1); + + // an existing directory without any pool file in it + o2::eventgen::EventPoolGenConfig emptyConfig; + emptyConfig.eventPoolPath = tmpDirGuard.path.string(); + o2::eventgen::GeneratorFromEventPool emptyPool(emptyConfig); + BOOST_CHECK_THROW(emptyPool.Init(), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("No file found that can be used with EventPool generator"), 2); +} + +/// with roundRobin the very same request is NOT fatal: the pool is started over +BOOST_AUTO_TEST_CASE(EventPool_RoundRobin_ExhaustionIsNotFatal) +{ + LogCatcher log("exhaustion_pool_rr", fair::Severity::warn); + TempDir tmpDirGuard("pool_rr"); + constexpr int numfiles = 2; + createPool(tmpDirGuard.path, numfiles); + int total = 0; + for (int i = 0; i < numfiles; ++i) { + total += eventsInFile(i); + } + + o2::eventgen::EventPoolGenConfig poolConfig; + poolConfig.eventPoolPath = tmpDirGuard.path.string(); + poolConfig.roundRobin = true; + poolConfig.randomize = false; + poolConfig.rngseed = 17; + o2::eventgen::GeneratorFromEventPool pool(poolConfig); + BOOST_REQUIRE(pool.Init()); + for (int i = 0; i < 3 * total; ++i) { + BOOST_CHECK_NO_THROW(BOOST_CHECK(readNextEvent(pool) >= 0.)); + } + BOOST_CHECK_EQUAL(log.count("ran out of events"), 0); +} + +/// the "input too small" estimate at initialisation uses the expected number of events of +/// the generator: the one of the job by default, none when declared unknown (sub-generators +/// of a hybrid only serve a share of the job and must not complain about the whole of it) +BOOST_AUTO_TEST_CASE(Init_WarnsOnlyForKnownExpectation) +{ + TempDir tmpDirGuard("init_warning"); + auto filenames = createPool(tmpDirGuard.path, 1); // 2 events in total + unsigned int total = 10; + o2::eventgen::Generator::setTotalNEvents(total); + { + LogCatcher log("init_warning_default", fair::Severity::warn); + o2::eventgen::GeneratorFromO2Kine gen(filenames); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_EQUAL(log.count("holds only about 2"), 1); + } + { + LogCatcher log("init_warning_unknown", fair::Severity::warn); + o2::eventgen::GeneratorFromO2Kine gen(filenames); + gen.setExpectedNEvents(0); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_EQUAL(log.count("holds only about"), 0); + } + { + // an expectation which fits in the pool does not warn either, whatever the job asks for + LogCatcher log("init_warning_fits", fair::Severity::warn); + o2::eventgen::GeneratorFromO2Kine gen(filenames); + gen.setExpectedNEvents(2); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_EQUAL(log.count("holds only about"), 0); + } + total = 0; + o2::eventgen::Generator::setTotalNEvents(total); +} + +/// the same single-file setup, but with roundRobin enabled: the already-open file must +/// be reused in place (no reopen), giving a fresh pass of the very same events +BOOST_AUTO_TEST_CASE(SingleFile_RoundRobin_ReusesWithoutReopening) +{ + TempDir tmpDirGuard("single_file_rr"); + auto const& tmpDir = tmpDirGuard.path; + auto file = (tmpDir / "kine.root").string(); + constexpr int nevents = 3; + createKineFile(file, nevents, 0); + + auto filesOpenBefore = openRootFiles(); + + o2::eventgen::O2KineGenConfig config; + config.roundRobin = true; + o2::eventgen::GeneratorFromO2Kine gen(config, {file}); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + + std::vector firstPass; + for (int i = 0; i < nevents; ++i) { + firstPass.push_back(readNextEvent(gen)); + } + // wrapping around must not close and reopen the file + for (int i = 0; i < nevents; ++i) { + BOOST_CHECK_CLOSE(readNextEvent(gen), firstPass[i], 1E-6); + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + } + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), 1); +} + +/// round robin must serve every event of every file, in order, and only then start +/// over with the first file again - for an arbitrary number of passes +BOOST_AUTO_TEST_CASE(Rollover_RoundRobin_FullPasses) +{ + TempDir tmpDirGuard("rollover_rr_full"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 3; + auto filenames = createPool(tmpDir, numfiles); + + auto filesOpenBefore = openRootFiles(); + + o2::eventgen::O2KineGenConfig config; + config.roundRobin = true; + o2::eventgen::GeneratorFromO2Kine gen(config, filenames); + BOOST_CHECK(gen.Init()); + + constexpr int npasses = 3; + for (int pass = 0; pass < npasses; ++pass) { + for (int file = 0; file < numfiles; ++file) { + for (int ev = 0; ev < eventsInFile(file); ++ev) { + // the very same sequence must come back on every pass + BOOST_CHECK_CLOSE(readNextEvent(gen), encodeMomentum(file, ev), 1E-6); + BOOST_CHECK_EQUAL(gen.getCurrentFileIndex(), file); + BOOST_CHECK_EQUAL(gen.getEventsAvailable(), eventsInFile(file)); + // laziness is not given up when wrapping around + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + } + } + } + // every file was opened exactly once per pass + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), numfiles * npasses); +} + +/// with randomization each round robin pass must again contain every event exactly +/// once, but in a freshly drawn order +BOOST_AUTO_TEST_CASE(Rollover_RoundRobin_RandomizedPasses) +{ + TempDir tmpDirGuard("rollover_rr_random"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 3; + auto filenames = createPool(tmpDir, numfiles); + + int total = 0; + std::set allEvents; + for (int file = 0; file < numfiles; ++file) { + total += eventsInFile(file); + for (int ev = 0; ev < eventsInFile(file); ++ev) { + allEvents.insert(encodeMomentum(file, ev)); + } + } + + o2::eventgen::O2KineGenConfig config; + config.roundRobin = true; + config.randomize = true; + config.rngseed = 99; + o2::eventgen::GeneratorFromO2Kine gen(config, filenames); + BOOST_CHECK(gen.Init()); + + constexpr int npasses = 4; + std::set> passOrders; + for (int pass = 0; pass < npasses; ++pass) { + std::set seen; + std::vector order; + for (int i = 0; i < total; ++i) { + auto id = readNextEvent(gen); + BOOST_CHECK(seen.insert(id).second); // no event twice within a pass + order.push_back(id); + } + // a full pass covers the whole input, no more and no less + BOOST_CHECK(seen == allEvents); + passOrders.insert(order); + } + // the passes must not all come out in the very same order + BOOST_CHECK(passOrders.size() > 1); +} + +/// the generator knows, through Generator::gTotalNEvents, how many events the job is +/// going to ask for, and counts how many it has actually served +BOOST_AUTO_TEST_CASE(Rollover_AccountsForRequestedEvents) +{ + TempDir tmpDirGuard("rollover_accounting"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 3; + auto filenames = createPool(tmpDir, numfiles); + int total = 0; + for (int i = 0; i < numfiles; ++i) { + total += eventsInFile(i); + } + + // this is what o2-sim / o2-sim-dpl-eventgen do before creating the generators + unsigned int requested = total; + o2::eventgen::Generator::setTotalNEvents(requested); + BOOST_CHECK_EQUAL(o2::eventgen::Generator::getTotalNEvents(), static_cast(total)); + + o2::eventgen::GeneratorFromO2Kine gen(filenames); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_EQUAL(gen.getEventsServed(), 0); + + for (int i = 0; i < total; ++i) { + BOOST_CHECK(readNextEvent(gen) >= 0.); + // the counter runs over the whole input, not per file + BOOST_CHECK_EQUAL(gen.getEventsServed(), i + 1); + } + BOOST_CHECK_EQUAL(gen.getEventsServed(), total); + + unsigned int reset = 0; + o2::eventgen::Generator::setTotalNEvents(reset); +} + +/// the event pool generator goes through the whole pool, one file after the other +BOOST_AUTO_TEST_CASE(EventPool_RollsOverAllFiles) +{ + TempDir tmpDirGuard("evtpool_rollover"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 5; + createPool(tmpDir, numfiles); + + int expectedEvents = 0; + for (int i = 0; i < numfiles; ++i) { + expectedEvents += eventsInFile(i); + } + + auto filesOpenBefore = openRootFiles(); + + o2::eventgen::EventPoolGenConfig config; + config.eventPoolPath = tmpDir.string(); + config.randomize = false; + config.rngseed = 42; + o2::eventgen::GeneratorFromEventPool gen(config); + BOOST_CHECK(gen.Init()); + BOOST_CHECK_EQUAL(gen.getFileUniverse().size(), static_cast(numfiles)); + BOOST_CHECK_EQUAL(gen.getChosenFiles().size(), static_cast(numfiles)); + BOOST_CHECK_EQUAL(gen.getO2KineGenerator()->getNumberOfFiles(), numfiles); + // still only one file open, whatever the size of the pool + BOOST_CHECK_EQUAL(gen.getO2KineGenerator()->getNumberOfFilesUsed(), 1); + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + + // every single event of the pool must be delivered exactly once + std::set seen; + for (int i = 0; i < expectedEvents; ++i) { + auto id = readNextEvent(gen); + BOOST_CHECK(id >= 0.); + BOOST_CHECK(seen.insert(id).second); // no duplicates + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + } + BOOST_CHECK_EQUAL(seen.size(), static_cast(expectedEvents)); + BOOST_CHECK_EQUAL(gen.getO2KineGenerator()->getNumberOfFilesUsed(), numfiles); +} + +/// the pool can be given as a text file listing the pool files (one per line): all the listed files, and +/// only those, are chained +BOOST_AUTO_TEST_CASE(EventPool_FileListChainsListedFiles) +{ + TempDir tmpDirGuard("evtpool_filelist"); + auto const& tmpDir = tmpDirGuard.path; + auto files = createPool(tmpDir, 4); + // leave out the file with index 1 + const std::vector listed = {0, 2, 3}; + auto listPath = tmpDir / "pools.txt"; + { + std::ofstream list(listPath); + for (auto i : listed) { + list << files[i] << "\n"; + } + } + int expectedEvents = 0; + std::set expectedIds; + for (auto i : listed) { + expectedEvents += eventsInFile(i); + for (int ev = 0; ev < eventsInFile(i); ++ev) { + expectedIds.insert(encodeMomentum(i, ev)); + } + } + + o2::eventgen::EventPoolGenConfig config; + config.eventPoolPath = listPath.string(); + config.randomize = false; + config.rngseed = 7; + o2::eventgen::GeneratorFromEventPool gen(config); + BOOST_REQUIRE(gen.Init()); + BOOST_CHECK_EQUAL(gen.getFileUniverse().size(), listed.size()); + BOOST_CHECK_EQUAL(gen.getO2KineGenerator()->getNumberOfFiles(), static_cast(listed.size())); + + std::set seen; + for (int i = 0; i < expectedEvents; ++i) { + BOOST_CHECK(seen.insert(readNextEvent(gen)).second); // no duplicates + } + BOOST_CHECK(seen == expectedIds); + BOOST_CHECK_EQUAL(gen.getO2KineGenerator()->getNumberOfFilesUsed(), static_cast(listed.size())); +} + +/// the order in which the pool files are visited must be reproducible for a given seed +BOOST_AUTO_TEST_CASE(EventPool_SelectionIsReproducible) +{ + TempDir tmpDirGuard("evtpool_selection"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 8; + createPool(tmpDir, numfiles); + + auto orderFor = [&tmpDir](unsigned int seed) { + o2::eventgen::EventPoolGenConfig config; + config.eventPoolPath = tmpDir.string(); + config.rngseed = seed; + o2::eventgen::GeneratorFromEventPool gen(config); + gen.Init(); + return gen.getChosenFiles(); + }; + + // the same seed always gives the same order, and every file is included + auto a = orderFor(1); + auto b = orderFor(1); + BOOST_CHECK_EQUAL(a.size(), static_cast(numfiles)); + BOOST_CHECK(a == b); + + // ... while different seeds do not all collapse onto the same order + std::set> orders; + for (unsigned int seed = 1; seed <= 10; ++seed) { + orders.insert(orderFor(seed)); + } + BOOST_CHECK(orders.size() > 1); +} + +/// for a given seed the order must not depend on the order in which the files are +/// found (a directory listing or alien find gives no fixed order) +BOOST_AUTO_TEST_CASE(EventPool_SelectionIndependentOfListingOrder) +{ + TempDir tmpDirGuard("evtpool_listorder"); + auto const& tmpDir = tmpDirGuard.path; + auto files = createPool(tmpDir, 8); + + auto orderFor = [&tmpDir](std::vector const& listing, std::string const& name) { + auto listPath = tmpDir / name; + { + std::ofstream list(listPath); + for (auto const& f : listing) { + list << f << "\n"; + } + } + o2::eventgen::EventPoolGenConfig config; + config.eventPoolPath = listPath.string(); + config.rngseed = 3; + o2::eventgen::GeneratorFromEventPool gen(config); + gen.Init(); + return gen.getChosenFiles(); + }; + + auto reversed = files; + std::reverse(reversed.begin(), reversed.end()); + auto a = orderFor(files, "pools_a.txt"); + auto b = orderFor(reversed, "pools_b.txt"); + BOOST_CHECK_EQUAL(a.size(), files.size()); + BOOST_CHECK(a == b); +} + +/// with randomization every event of the pool is still served exactly once: the order +/// within a file is a permutation of its entries, fixed when the file is opened +BOOST_AUTO_TEST_CASE(EventPool_RandomizeIsAPermutation) +{ + TempDir tmpDirGuard("evtpool_randomize"); + auto const& tmpDir = tmpDirGuard.path; + constexpr int numfiles = 4; + createPool(tmpDir, numfiles); + + int expectedEvents = 0; + for (int i = 0; i < numfiles; ++i) { + expectedEvents += eventsInFile(i); + } + + auto filesOpenBefore = openRootFiles(); + + o2::eventgen::EventPoolGenConfig config; + config.eventPoolPath = tmpDir.string(); + config.randomize = true; + config.rngseed = 12345; + o2::eventgen::GeneratorFromEventPool gen(config); + BOOST_CHECK(gen.Init()); + + std::set seen; + std::vector order; + for (int i = 0; i < expectedEvents; ++i) { + auto id = readNextEvent(gen); + BOOST_CHECK(id >= 0.); + // no event is served twice ... + BOOST_CHECK(seen.insert(id).second); + order.push_back(id); + // ... and still only one file is open + BOOST_CHECK_EQUAL(openRootFiles() - filesOpenBefore, 1); + } + BOOST_CHECK_EQUAL(seen.size(), static_cast(expectedEvents)); + BOOST_CHECK_EQUAL(gen.getO2KineGenerator()->getNumberOfFilesUsed(), numfiles); + + // the order must actually differ from the sequential one + auto sorted = order; + std::sort(sorted.begin(), sorted.end()); + BOOST_CHECK(order != sorted); +} + +/// randomization is a permutation also for a single file, and round robin gives a +/// fresh permutation on every pass +BOOST_AUTO_TEST_CASE(SingleFile_RandomizeRoundRobin) +{ + TempDir tmpDirGuard("single_randomize"); + auto const& tmpDir = tmpDirGuard.path; + auto file = (tmpDir / "kine.root").string(); + constexpr int nevents = 6; + createKineFile(file, nevents, 0); + + o2::eventgen::O2KineGenConfig config; + config.randomize = true; + config.roundRobin = true; + config.rngseed = 7; + o2::eventgen::GeneratorFromO2Kine gen(config, {file}); + BOOST_CHECK(gen.Init()); + + // first pass: every event exactly once + std::set firstPass; + std::vector firstOrder; + for (int i = 0; i < nevents; ++i) { + auto id = readNextEvent(gen); + BOOST_CHECK(firstPass.insert(id).second); + firstOrder.push_back(id); + } + BOOST_CHECK_EQUAL(firstPass.size(), static_cast(nevents)); + + // second pass: same events again, but re-shuffled and without reopening the file + std::set secondPass; + std::vector secondOrder; + for (int i = 0; i < nevents; ++i) { + auto id = readNextEvent(gen); + BOOST_CHECK(secondPass.insert(id).second); + secondOrder.push_back(id); + } + BOOST_CHECK(firstPass == secondPass); + BOOST_CHECK(firstOrder != secondOrder); + BOOST_CHECK_EQUAL(gen.getNumberOfFilesUsed(), 1); +} diff --git a/Generators/test/test_GeneratorHybridEventPool.cxx b/Generators/test/test_GeneratorHybridEventPool.cxx new file mode 100644 index 0000000000000..7468b43b9faa4 --- /dev/null +++ b/Generators/test/test_GeneratorHybridEventPool.cxx @@ -0,0 +1,1002 @@ +// Copyright 2019-2026 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +/// \file test_GeneratorHybridEventPool.cxx +/// \brief regression tests for the use of event pools (and finite inputs in general) inside the GeneratorHybrid +/// \author M. Giacalone, mgiacalo@cern.ch, 10/2026 +/// +/// The GeneratorHybrid is a process-wide singleton and the tests of the failure modes end the process, so +/// every scenario runs in a child process: this very test executable, started again with the scenario in the +/// environment (see HybridScenarioWorker at the end). + +#define BOOST_TEST_MODULE Test GeneratorHybridEventPool +#define BOOST_TEST_MAIN +#define BOOST_TEST_DYN_LINK +#include + +#include "EventPoolTestUtils.h" +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace evtpooltest; +using o2::eventgen::Generator; +using o2::eventgen::GeneratorFromEventPool; +using o2::eventgen::GeneratorHybrid; + +namespace +{ + +constexpr const char* kScenarioEnv = "O2_HYBRIDTEST_SCENARIO"; // scenario to run in the child process +constexpr const char* kDirEnv = "O2_HYBRIDTEST_DIR"; // scratch directory of the child process +constexpr const char* kWindowEnv = "O2_HYBRIDTEST_WINDOW_MS"; // stretching of the race window (0 = natural timing) +constexpr int kFatalExitCode = 86; // exit code of the child on LOG(fatal) (too_few_events, uneven_shares) +constexpr int kWatchdogExitCode = 87; // exit code of a child which did not end in time (e.g. a hang) +constexpr int kWatchdogSeconds = 180; +constexpr int kDefaultWindowMs = 300; + +// --------------------------------------------------------------------------------------------------------------- +// building of the configuration of the hybrid generator +// --------------------------------------------------------------------------------------------------------------- + +std::string boxJson(int pdg, int number) +{ + // the hybrid reads the configuration with TBufferJSON, which needs every member of BoxGenConfig + std::ostringstream s; + s << R"({ "name": "boxgen", "config": { "pdg": )" << pdg << R"(, "number": )" << number + << R"(, "eta": [-1, 1], "prange": [0.1, 5], "phirange": [0, 360], "sampleYAndPt": false } })"; + return s.str(); +} + +std::string poolJson(std::string const& path) +{ + std::ostringstream s; + s << R"({ "name": "evtpool", "config": { "eventPoolPath": ")" << path + << R"(", "skipNonTrackable": true, "roundRobin": false, "randomize": false, "rngseed": 1, "randomphi": false } })"; + return s.str(); +} + +/// a HepMC sub-generator reading one file; with randomize it serves every event of the file once, in random order +std::string hepmcJson(std::string const& file, bool randomize) +{ + // as for the box generator, every member of FileOrCmdGenConfig and HepMCGenConfig must be given + std::ostringstream s; + s << R"({ "name": "hepmc", "config": { "configcmd": { "fileNames": ")" << file << R"(", "cmd": "" }, )" + << R"("confighepmc": { "version": 0, "eventsToSkip": 0, "fileName": "", "prune": false, "randomize": )" + << (randomize ? "true" : "false") + << R"(, "roundRobin": false, "reshuffleOnRepeat": false, "rngseed": 1 } } })"; + return s.str(); +} + +/// writes a HepMC3 ASCII file with nevents events; event i holds a proton beam particle and one K+ with px = i + 1 +void createHepMCFile(std::string const& path, int nevents) +{ + HepMC3::WriterAscii writer(path); + for (int i = 0; i < nevents; ++i) { + HepMC3::GenEvent event(HepMC3::Units::GEV, HepMC3::Units::MM); + event.set_event_number(i); + const double px = i + 1.; + const double mK = 0.493677; + auto beam = std::make_shared(HepMC3::FourVector(0., 0., 100., std::sqrt(100. * 100. + 0.938272 * 0.938272)), 2212, 4); + auto kaon = std::make_shared(HepMC3::FourVector(px, 0., 0., std::sqrt(px * px + mK * mK)), 321, 1); + auto vertex = std::make_shared(); + vertex->add_particle_in(beam); + vertex->add_particle_out(kaon); + event.add_vertex(vertex); + writer.write_event(event); + } + writer.close(); +} + +std::string cocktailJson(std::vector const& subs) +{ + std::string s = R"({ "cocktail": [ )"; + for (size_t i = 0; i < subs.size(); ++i) { + s += (i ? ", " : "") + subs[i]; + } + return s + " ] }"; +} + +std::string hybridJson(std::vector const& entries, std::string const& fractions, std::string const& mode = "") +{ + std::string s = "{\n"; + if (!mode.empty()) { + s += R"( "mode": ")" + mode + "\",\n"; + } + s += " \"generators\": [\n"; + for (size_t i = 0; i < entries.size(); ++i) { + s += " " + entries[i] + (i + 1 < entries.size() ? ",\n" : "\n"); + } + s += " ],\n \"fractions\": [ " + fractions + " ]\n}\n"; + return s; +} + +std::string writeFile(fs::path const& path, std::string const& content) +{ + fs::create_directories(path.parent_path()); + std::ofstream out(path); + out << content; + return path.string(); +} + +// --------------------------------------------------------------------------------------------------------------- +// helpers of the child process +// --------------------------------------------------------------------------------------------------------------- + +/// Stretches the time between the scheduling of the lookahead task of the last event and the raising of the stop +/// flag (see the file description). Does nothing if `ms` is 0. +class RaceWindowWidener +{ + public: + explicit RaceWindowWidener(int ms) : mMs(ms) + { + if (mMs > 0) { + fair::Logger::AddCustomSink("race_window_widener", fair::Severity::info, [this](std::string const& content, fair::LogMetaData const&) { + if (content.find("HybridGen: Stopping TBB task pool") != std::string::npos) { + mTriggered = true; + std::this_thread::sleep_for(std::chrono::milliseconds(mMs)); + } + }); + } + } + ~RaceWindowWidener() + { + if (mMs > 0) { + fair::Logger::RemoveCustomSink("race_window_widener"); + } + } + /// whether the stretched window was entered; if not (e.g. because the message above was reworded) the + /// scenarios would pass without actually testing anything + bool triggered() const { return mTriggered; } + bool enabled() const { return mMs > 0; } + + private: + int mMs; + std::atomic mTriggered{false}; +}; + +void configureHybrid(int numWorkers, bool randomize) +{ + o2::eventgen::GeneratorHybridParam::Instance(); // makes sure that the parameters are registered + o2::conf::ConfigurableParam::updateFromString("GeneratorHybrid.num_workers=" + std::to_string(numWorkers) + + ";GeneratorHybrid.randomize=" + (randomize ? "true" : "false")); +} + +/// what the job gets for one event +struct EventRecord { + std::vector particles; + std::string forwardingGenerator; // as written to the event header +}; + +/// runs the event loop of the job: the two calls are what the primary generator does for every event +std::vector produce(GeneratorHybrid& hybrid, unsigned int nevents) +{ + std::vector events; + for (unsigned int i = 0; i < nevents; ++i) { + BOOST_REQUIRE(hybrid.generateEvent()); + BOOST_REQUIRE(hybrid.importParticles()); + EventRecord record; + record.particles = hybrid.getParticles(); + o2::dataformats::MCEventHeader header; + hybrid.updateHeader(&header); + bool valid = false; + record.forwardingGenerator = header.getInfo("forwarding-generator", valid); + events.push_back(std::move(record)); + } + return events; +} + +/// gives the worker threads the time to do whatever they still want to do after the job has ended +void settle() { std::this_thread::sleep_for(std::chrono::milliseconds(300)); } + +/// identifier of the pool event (first track of the pool event; see encodeMomentum), -1 if the event has none +double poolEventId(EventRecord const& event) +{ + for (auto const& p : event.particles) { + if (p.GetPdgCode() == 211) { + return p.Px(); + } + } + return -1.; +} + +/// identifiers of all the events of a pool made with createPool(…, nfiles) +std::set allPoolEventIds(int nfiles) +{ + std::set ids; + for (int f = 0; f < nfiles; ++f) { + for (int e = 0; e < eventsInFile(f); ++e) { + ids.insert(encodeMomentum(f, e)); + } + } + return ids; +} + +int poolEventsInFiles(int nfiles) +{ + int n = 0; + for (int f = 0; f < nfiles; ++f) { + n += eventsInFile(f); + } + return n; +} + +GeneratorFromEventPool* subPool(GeneratorHybrid& hybrid, size_t index) +{ + auto const& gens = hybrid.getGenerators(); + BOOST_REQUIRE(index < gens.size()); + auto pool = dynamic_cast(gens[index].get()); + BOOST_REQUIRE(pool != nullptr); + return pool; +} + +struct ScenarioEnv { + fs::path dir; + int windowMs = kDefaultWindowMs; +}; + +/// the number of events each sub-generator of the hybrid was told to serve +std::vector expectedShares(GeneratorHybrid& hybrid) +{ + std::vector shares; + for (auto const& gen : hybrid.getGenerators()) { + shares.push_back(gen->getExpectedNEvents()); + } + return shares; +} + +/// common start of the scenarios: configures the generator and creates the singleton +GeneratorHybrid& startHybrid(ScenarioEnv const& env, std::string const& json, unsigned int nevents, int numWorkers = 2, bool randomize = false) +{ + auto jsonPath = writeFile(env.dir / "hybrid.json", json); + // the (lazily built, not thread-safe) particle database must be complete before several worker threads start to + // use it; Instance() also marks it as initialised, so that the generators calling O2DatabasePDG::Instance() in the + // worker threads do not add the ALICE particles again + o2::O2DatabasePDG::Instance(); + configureHybrid(numWorkers, randomize); + static unsigned int total; // setTotalNEvents wants an lvalue + total = nevents; + Generator::setTotalNEvents(total); + return GeneratorHybrid::Instance(jsonPath); +} + +// --------------------------------------------------------------------------------------------------------------- +// the scenarios; each one runs in a process of its own +// --------------------------------------------------------------------------------------------------------------- + +/// The job asks for exactly as many events as the pool holds (9 events in 3 files). Every one of them must be +/// served exactly once, and the pool must not be asked for another one after the last event of the job. +void scenarioExactPool(ScenarioEnv const& env) +{ + LogCatcher log("scenario_exact_pool"); // warnings and worse; LOG(fatal) throws + RaceWindowWidener widener(env.windowMs); + constexpr int nfiles = 3; + auto poolDir = env.dir / "pool"; + createPool(poolDir, nfiles); + const unsigned int nevents = poolEventsInFiles(nfiles); + auto& hybrid = startHybrid(env, hybridJson({poolJson(poolDir.string())}, "1"), nevents); + BOOST_REQUIRE(hybrid.Init()); + // the only generator serves all the events, and the pool passes this on to the generator reading the files + BOOST_CHECK((expectedShares(hybrid) == std::vector{nevents})); + BOOST_CHECK_EQUAL(subPool(hybrid, 0)->getO2KineGenerator()->getExpectedNEvents(), nevents); + BOOST_CHECK_EQUAL(log.count("holds only about"), 0); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + std::set ids; + for (auto const& e : events) { + ids.insert(poolEventId(e)); + BOOST_CHECK_EQUAL(e.forwardingGenerator, "HybridGen"); + } + BOOST_CHECK(ids == allPoolEventIds(nfiles)); // every event of the pool exactly once + auto pool = subPool(hybrid, 0); + std::cout << "HYBRIDTEST served=" << pool->getO2KineGenerator()->getEventsServed() << " requested=" << nevents << std::endl; + BOOST_CHECK_EQUAL(pool->getO2KineGenerator()->getEventsServed(), static_cast(nevents)); + BOOST_CHECK_EQUAL(pool->getO2KineGenerator()->getNumberOfFilesUsed(), nfiles); + BOOST_CHECK_EQUAL(log.count("ran out of events"), 0); + BOOST_CHECK_MESSAGE(!widener.enabled() || widener.triggered(), "the race window was not entered: the scenario tests nothing"); +} + +/// The pool holds more events than the job asks for (14 against 9). The pool must have delivered exactly the +/// 9 events which were consumed: with the lookahead after the last event it delivers 10. This is the same +/// check as above, but it does not depend on the fatal error of a pool which is used up. With randomize the +/// shares are not exact, there is no request budget, and only the guard on the last event of the job helps. +void scenarioLargerPool(ScenarioEnv const& env, bool randomize) +{ + LogCatcher log("scenario_larger_pool"); + RaceWindowWidener widener(env.windowMs); + constexpr int nfiles = 4; + auto poolDir = env.dir / "pool"; + createPool(poolDir, nfiles); + const unsigned int nevents = 9; + auto& hybrid = startHybrid(env, hybridJson({poolJson(poolDir.string())}, "1"), nevents, 2, randomize); + BOOST_REQUIRE(hybrid.Init()); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + std::set ids; + for (auto const& e : events) { + ids.insert(poolEventId(e)); + } + BOOST_CHECK_EQUAL(ids.size(), static_cast(nevents)); // no event twice + auto pool = subPool(hybrid, 0); + std::cout << "HYBRIDTEST served=" << pool->getO2KineGenerator()->getEventsServed() << " requested=" << nevents << std::endl; + BOOST_CHECK_EQUAL(pool->getO2KineGenerator()->getEventsServed(), static_cast(nevents)); + BOOST_CHECK_MESSAGE(!widener.enabled() || widener.triggered(), "the race window was not entered: the scenario tests nothing"); +} + +/// Cocktail: every event merges a box generator (2 muons) and an event pool (a chain of tracks), the pool +/// holds exactly as many events as the job asks for. Checks the merging (order, mother/daughter indices), +/// the single use of the pool events and - as for the plain mode - that nothing is scheduled after the last +/// event of the job, which in the cocktail code is a place of its own. +void scenarioCocktailPool(ScenarioEnv const& env) +{ + LogCatcher log("scenario_cocktail_pool"); + RaceWindowWidener widener(env.windowMs); + constexpr int nfiles = 3; + constexpr int nbox = 2; + auto poolDir = env.dir / "pool"; + createPool(poolDir, nfiles, /*chain=*/true); + const unsigned int nevents = poolEventsInFiles(nfiles); + auto json = hybridJson({cocktailJson({boxJson(13, nbox), poolJson(poolDir.string())})}, "1"); + auto& hybrid = startHybrid(env, json, nevents); + BOOST_REQUIRE(hybrid.Init()); + // every member of the cocktail contributes to every event + BOOST_CHECK((expectedShares(hybrid) == std::vector{nevents, nevents})); + BOOST_CHECK_EQUAL(log.count("holds only about"), 0); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + std::set ids; + for (auto const& e : events) { + auto const& ps = e.particles; + BOOST_REQUIRE_GE(ps.size(), static_cast(nbox + 1)); + // the particles of the first generator of the cocktail come first, then the ones of the pool + for (int i = 0; i < nbox; ++i) { + BOOST_CHECK_EQUAL(ps[i].GetPdgCode(), 13); + BOOST_CHECK_EQUAL(ps[i].GetMother(0), -1); + } + const int npool = ps.size() - nbox; + ids.insert(ps[nbox].Px()); + for (int j = 0; j < npool; ++j) { + auto const& p = ps[nbox + j]; + BOOST_CHECK_EQUAL(p.GetPdgCode(), 211); + BOOST_CHECK_CLOSE(p.Px(), ps[nbox].Px(), 1E-6); // all tracks of a pool event carry its identifier + // the chain of the pool event is shifted by the particles which came before it + BOOST_CHECK_EQUAL(p.GetMother(0), j > 0 ? nbox + j - 1 : -1); + BOOST_CHECK_EQUAL(p.GetDaughter(0), j < npool - 1 ? nbox + j + 1 : -1); + } + BOOST_CHECK_EQUAL(e.forwardingGenerator, "HybridGen"); + } + BOOST_CHECK(ids == allPoolEventIds(nfiles)); + auto pool = subPool(hybrid, 1); + std::cout << "HYBRIDTEST served=" << pool->getO2KineGenerator()->getEventsServed() << " requested=" << nevents << std::endl; + BOOST_CHECK_EQUAL(pool->getO2KineGenerator()->getEventsServed(), static_cast(nevents)); + BOOST_CHECK_EQUAL(log.count("ran out of events"), 0); + BOOST_CHECK_MESSAGE(!widener.enabled() || widener.triggered(), "the race window was not entered: the scenario tests nothing"); +} + +/// Cocktail and plain generators next to each other: group 0 is the cocktail of three box generators (2 muons, +/// 1 pion), group 1 is a single photon; the fractions make the two alternate. +void scenarioCocktailGroups(ScenarioEnv const& env) +{ + LogCatcher log("scenario_cocktail_groups"); + RaceWindowWidener widener(env.windowMs); + const unsigned int nevents = 6; + auto json = hybridJson({cocktailJson({boxJson(13, 2), boxJson(211, 1)}), boxJson(22, 1)}, "1, 1"); + auto& hybrid = startHybrid(env, json, nevents); + BOOST_REQUIRE(hybrid.Init()); + // the fraction is the one of the group, it applies to every member of it + BOOST_CHECK((expectedShares(hybrid) == std::vector{3, 3, 3})); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + for (unsigned int i = 0; i < nevents; ++i) { + std::vector pdgs; + for (auto const& p : events[i].particles) { + pdgs.push_back(p.GetPdgCode()); + } + std::sort(pdgs.begin(), pdgs.end()); + if (i % 2 == 0) { + BOOST_CHECK((pdgs == std::vector{13, 13, 211})); + } else { + BOOST_CHECK((pdgs == std::vector{22})); + } + } + BOOST_CHECK_MESSAGE(!widener.enabled() || widener.triggered(), "the race window was not entered: the scenario tests nothing"); +} + +/// Parallel mode: clones of one generator share the work +void scenarioParallelMode(ScenarioEnv const& env) +{ + LogCatcher log("scenario_parallel_mode"); + RaceWindowWidener widener(env.windowMs); + const unsigned int nevents = 7; + auto json = hybridJson({boxJson(13, 1), boxJson(13, 1), boxJson(13, 1)}, "1, 1, 1", "parallel"); + auto& hybrid = startHybrid(env, json, nevents, 3); + BOOST_REQUIRE(hybrid.Init()); + // any clone can serve any event: the share of each one is not known + BOOST_CHECK((expectedShares(hybrid) == std::vector{0, 0, 0})); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + for (auto const& e : events) { + BOOST_REQUIRE_EQUAL(e.particles.size(), 1u); + BOOST_CHECK_EQUAL(e.particles[0].GetPdgCode(), 13); + } + BOOST_CHECK_MESSAGE(!widener.enabled() || widener.triggered(), "the race window was not entered: the scenario tests nothing"); +} + +/// An event pool next to a box generator, the job asks for more events (10) than the pool holds (9): the pool +/// only serves its share (5). The pool must be told its share and not the number of events of the whole job, +/// i.e. it must not warn that its input is too small. +void scenarioMixedSequence(ScenarioEnv const& env) +{ + LogCatcher log("scenario_mixed_sequence"); + RaceWindowWidener widener(env.windowMs); + auto poolFile = (env.dir / "pool" / "evtpool.root").string(); + fs::create_directories(env.dir / "pool"); + createKineFile(poolFile, 9, 0); + const unsigned int nevents = 10; + auto& hybrid = startHybrid(env, hybridJson({poolJson(poolFile), boxJson(13, 1)}, "1, 1"), nevents); + BOOST_REQUIRE(hybrid.Init()); + + // fractions 1:1 over 10 events + BOOST_CHECK((expectedShares(hybrid) == std::vector{5, 5})); + BOOST_CHECK_EQUAL(subPool(hybrid, 0)->getO2KineGenerator()->getExpectedNEvents(), 5u); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + int fromPool = 0; + std::set ids; + for (auto const& e : events) { + if (poolEventId(e) >= 0.) { + fromPool++; + ids.insert(poolEventId(e)); + } + } + BOOST_CHECK_EQUAL(fromPool, 5); + BOOST_CHECK_EQUAL(ids.size(), 5u); // the 5 events are different ones + BOOST_CHECK_EQUAL(log.count("holds only about"), 0); + BOOST_CHECK_EQUAL(log.count("ran out of events"), 0); + BOOST_CHECK_MESSAGE(!widener.enabled() || widener.triggered(), "the race window was not entered: the scenario tests nothing"); +} + +/// Fractions 2:1 over 10 events: the generators are used in turn, 2 events from the pool, then 1 from the +/// box, so the pool serves 7 events (2+2+2 and the first of the last cycle) and the box 3. A pool holding 6 +/// events is too small for this share: the pool says so at initialisation, and the job stops with +/// 'ran out of events', reporting the share of the pool. The child ends with kFatalExitCode, the parent checks +/// its output. +void scenarioUnevenShares(ScenarioEnv const& env) +{ + fair::Logger::OnFatal([] { + std::cout.flush(); + std::cerr.flush(); + std::_Exit(kFatalExitCode); + }); + auto poolFile = (env.dir / "pool" / "evtpool.root").string(); + fs::create_directories(env.dir / "pool"); + createKineFile(poolFile, 6, 0); + auto& hybrid = startHybrid(env, hybridJson({poolJson(poolFile), boxJson(13, 1)}, "2, 1"), 10); + BOOST_REQUIRE(hybrid.Init()); + auto shares = expectedShares(hybrid); + std::cout << "HYBRIDTEST shares="; + for (auto n : shares) { + std::cout << n << ","; + } + std::cout << " kine=" << subPool(hybrid, 0)->getO2KineGenerator()->getExpectedNEvents() << std::endl; + produce(hybrid, 10); // does not come back + BOOST_ERROR("the job went through although the pool holds fewer events than its share"); +} + +/// With randomize the generator of each event is drawn with probability fraction/sum: the share is not known +/// in advance, the mean (rounded up) is used. Fractions 3:1 over 10 events: 7.5 and 2.5, i.e. 8 and 3. +void scenarioRandomShares(ScenarioEnv const& env) +{ + LogCatcher log("scenario_random_shares"); + const unsigned int nevents = 10; + auto& hybrid = startHybrid(env, hybridJson({boxJson(13, 1), boxJson(22, 1)}, "3, 1"), nevents, 2, /*randomize=*/true); + BOOST_REQUIRE(hybrid.Init()); + BOOST_CHECK((expectedShares(hybrid) == std::vector{8, 3})); + + // run the job to its end, so that the worker threads stop before the generator is destroyed + auto events = produce(hybrid, nevents); + settle(); + BOOST_CHECK_EQUAL(events.size(), nevents); +} + +/// An event pool which does not exist stops the job at initialisation, instead of being used uninitialised later +/// on. The pool says so itself (as it does outside a hybrid), before the hybrid checks the result of its Init(). +void scenarioInitFailure(ScenarioEnv const& env) +{ + LogCatcher log("scenario_init_failure", fair::Severity::error); + auto& hybrid = startHybrid(env, hybridJson({poolJson((env.dir / "does_not_exist").string())}, "1"), 5); + BOOST_CHECK_THROW(hybrid.Init(), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("No file found that can be used with EventPool generator"), 1); +} + +/// A sub-generator whose Init() only reports a failure through its result (here: GeneratorHepMC with an input +/// file which does not exist) stops the job at initialisation as well: the hybrid checks that result. +void scenarioInitFailureHepMC(ScenarioEnv const& env) +{ + LogCatcher log("scenario_init_failure_hepmc", fair::Severity::error); + auto json = hybridJson({hepmcJson((env.dir / "does_not_exist.hepmc").string(), false), boxJson(13, 1)}, "1, 1"); + auto& hybrid = startHybrid(env, json, 5); + BOOST_CHECK_THROW(hybrid.Init(), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("Initialization of sub-generator 0 (hepmc) failed"), 1); +} + +/// A HepMC file (5 events, served in random order, each once) next to a box generator, fractions 1:1 over 10 +/// events: the HepMC generator only serves 5 events. It must be told this share, and not the 10 events of the +/// job: no "input too small" warning, and, as for the pools, it is not asked for a 6th event. +void scenarioHepMCShare(ScenarioEnv const& env) +{ + LogCatcher log("scenario_hepmc_share"); + auto hepmcFile = (env.dir / "events.hepmc").string(); + fs::create_directories(env.dir); + createHepMCFile(hepmcFile, 5); + const unsigned int nevents = 10; + auto& hybrid = startHybrid(env, hybridJson({hepmcJson(hepmcFile, true), boxJson(13, 1)}, "1, 1"), nevents); + BOOST_REQUIRE(hybrid.Init()); + BOOST_CHECK((expectedShares(hybrid) == std::vector{5, 5})); + BOOST_CHECK_EQUAL(log.count("This job will request"), 0); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + std::set kaons; + int fromBox = 0; + for (auto const& e : events) { + for (auto const& p : e.particles) { + if (p.GetPdgCode() == 321) { + kaons.insert(std::round(p.Px() * 1000.) / 1000.); + } else if (p.GetPdgCode() == 13) { + fromBox++; + } + } + } + std::cout << "HYBRIDTEST hepmc_events=" << kaons.size() << " box_events=" << fromBox << std::endl; + BOOST_CHECK((kaons == std::set{1., 2., 3., 4., 5.})); // every HepMC event exactly once + BOOST_CHECK_EQUAL(fromBox, 5); + BOOST_CHECK_EQUAL(log.count("ran out of events"), 0); +} + +/// A negative fraction is rejected when the configuration is read: the sequential mode would stay on that +/// generator for ever (its event counter never equals a negative fraction) +void scenarioNegativeFraction(ScenarioEnv const& env) +{ + LogCatcher log("scenario_negative_fraction", fair::Severity::error); + BOOST_CHECK_THROW(startHybrid(env, hybridJson({boxJson(13, 1), boxJson(22, 1)}, "-1, 1"), 5), fair::FatalException); + BOOST_CHECK_EQUAL(log.count("Fractions must be non-negative integers"), 1); +} + +/// All fractions 0 is rejected also in random mode, where the first generator would silently be used for every +/// event (no probability is assigned to any of them) +void scenarioZeroFractionsRandom(ScenarioEnv const& env) +{ + LogCatcher log("scenario_zero_fractions_random", fair::Severity::error); + BOOST_CHECK_THROW(startHybrid(env, hybridJson({boxJson(13, 1), boxJson(22, 1)}, "0, 0"), 5, 2, /*randomize=*/true), + fair::FatalException); + BOOST_CHECK_EQUAL(log.count("All fractions provided are 0"), 1); +} + +/// The job asks for one event more (10) than the pool holds (9): this must remain a fatal error - the job must +/// not end "successfully" with fewer events than requested. The LOG(fatal) comes from a worker thread, where it +/// cannot be turned into an exception which could be caught: the process ends with the exit code below. +void scenarioTooFewEvents(ScenarioEnv const& env) +{ + fair::Logger::OnFatal([] { + std::cout.flush(); + std::cerr.flush(); + std::_Exit(kFatalExitCode); + }); + constexpr int nfiles = 3; + auto poolDir = env.dir / "pool"; + createPool(poolDir, nfiles); + const unsigned int nevents = poolEventsInFiles(nfiles) + 1; + auto& hybrid = startHybrid(env, hybridJson({poolJson(poolDir.string())}, "1"), nevents); + BOOST_REQUIRE(hybrid.Init()); + produce(hybrid, nevents); // does not come back + BOOST_ERROR("the job went through although the pool holds fewer events than requested"); +} + +/// Fractions 9:9 over 18 events: the pool (9 events) serves the first 9 events of the job, the box the last 9. +/// The pool is used up in the middle of the job, so the guard on the last event of the job does not help: the +/// request budget must keep the hybrid from asking the pool for a 10th event while the box events are produced. +void scenarioMidJobShare(ScenarioEnv const& env) +{ + LogCatcher log("scenario_mid_job_share"); + auto poolFile = (env.dir / "pool" / "evtpool.root").string(); + fs::create_directories(env.dir / "pool"); + createKineFile(poolFile, 9, 0); + const unsigned int nevents = 18; + auto& hybrid = startHybrid(env, hybridJson({poolJson(poolFile), boxJson(13, 1)}, "9, 9"), nevents); + BOOST_REQUIRE(hybrid.Init()); + BOOST_CHECK((expectedShares(hybrid) == std::vector{9, 9})); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + for (unsigned int i = 0; i < nevents; ++i) { + BOOST_CHECK_EQUAL(poolEventId(events[i]) >= 0., i < 9); + } + auto pool = subPool(hybrid, 0); + std::cout << "HYBRIDTEST served=" << pool->getO2KineGenerator()->getEventsServed() << std::endl; + BOOST_CHECK_EQUAL(pool->getO2KineGenerator()->getEventsServed(), 9); + BOOST_CHECK_EQUAL(log.count("ran out of events"), 0); +} + +/// Fractions 0:1: the pool is never used, so it is not asked for any event at all (it used to produce one in vain) +void scenarioZeroFraction(ScenarioEnv const& env) +{ + LogCatcher log("scenario_zero_fraction"); + auto poolFile = (env.dir / "pool" / "evtpool.root").string(); + fs::create_directories(env.dir / "pool"); + createKineFile(poolFile, 9, 0); + const unsigned int nevents = 5; + auto& hybrid = startHybrid(env, hybridJson({poolJson(poolFile), boxJson(13, 1)}, "0, 1"), nevents); + BOOST_REQUIRE(hybrid.Init()); + BOOST_CHECK((expectedShares(hybrid) == std::vector{0, 5})); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + for (auto const& e : events) { + BOOST_CHECK_LT(poolEventId(e), 0.); + } + BOOST_CHECK_EQUAL(subPool(hybrid, 0)->getO2KineGenerator()->getEventsServed(), 0); +} + +/// Random mode, two pools with fractions 1:1: pool A holds 9 events, pool B enough for the job. The seed is chosen +/// such that A is drawn exactly 9 times and the event after the 9th one comes from B: the lookahead asks A for a +/// 10th event, which fails in the worker thread. This event is never consumed, so the job must go through. Only +/// the main thread draws from gRandom here (the pools neither shuffle nor rotate), so the sequence is reproducible. +void scenarioUnusedFailure(ScenarioEnv const& env) +{ + LogCatcher log("scenario_unused_failure"); // LOG(fatal) throws, as in the simulation devices + // keeps the job alive long enough after its last event for a worker to pick up the lookahead task of pool A + RaceWindowWidener widener(env.windowMs); + auto poolFileA = (env.dir / "poolA" / "evtpool.root").string(); + fs::create_directories(env.dir / "poolA"); + createKineFile(poolFileA, 9, 0); + auto poolDirB = env.dir / "poolB"; + createPool(poolDirB, 5); // 20 events + + // the generator of each event is pool A if the draw is <= 0.5 (see GeneratorHybrid::generateEvent) + constexpr int nA = 9; + unsigned int nevents = 0; + UInt_t seed = 0; + while (nevents == 0) { + gRandom->SetSeed(++seed); + int a = 0; + for (unsigned int i = 0; i < 25; ++i) { // at most 17 events from pool B + if (gRandom->Rndm() <= 0.5 && ++a == nA) { + if (gRandom->Rndm() > 0.5) { + nevents = i + 2; // the 9th event of A, then one of B, which ends the job + } + break; + } + } + } + std::cout << "HYBRIDTEST seed=" << seed << " nevents=" << nevents << std::endl; + auto json = hybridJson({poolJson(poolFileA), poolJson(poolDirB.string())}, "1, 1"); + auto& hybrid = startHybrid(env, json, nevents, 2, /*randomize=*/true); + BOOST_REQUIRE(hybrid.Init()); + gRandom->SetSeed(seed); + + auto events = produce(hybrid, nevents); + settle(); + + BOOST_REQUIRE_EQUAL(events.size(), nevents); + BOOST_CHECK_EQUAL(subPool(hybrid, 0)->getO2KineGenerator()->getEventsServed(), nA); + // the lookahead did fail, but this did not matter + BOOST_CHECK_EQUAL(log.count("ran out of events after 9 event(s)"), 1); + BOOST_CHECK_EQUAL(log.count("failed to generate the requested event"), 0); + BOOST_CHECK_MESSAGE(!widener.enabled() || widener.triggered(), "the race window was not entered: the scenario tests nothing"); +} + +/// As too_few_events, but LOG(fatal) throws (as in the simulation devices) and two worker threads run: the +/// failure of the pool in the worker thread must reach the main thread. Without the forwarding the exception +/// stays in the TBB worker, the other worker keeps polling and the job hangs (the watchdog of the child ends it). +void scenarioFailureForwarded(ScenarioEnv const& env) +{ + LogCatcher log("scenario_failure_forwarded", fair::Severity::error); + constexpr int nfiles = 3; + auto poolDir = env.dir / "pool"; + createPool(poolDir, nfiles); + const unsigned int nevents = poolEventsInFiles(nfiles) + 1; + auto& hybrid = startHybrid(env, hybridJson({poolJson(poolDir.string())}, "1"), nevents, 2); + BOOST_REQUIRE(hybrid.Init()); + BOOST_CHECK_THROW(produce(hybrid, nevents), fair::FatalException); + settle(); // the worker threads are detached: let them see the stop flag before the process ends + BOOST_CHECK_EQUAL(log.count("ran out of events after 9 event(s) from 3 input file(s) (10 were requested)"), 1); + BOOST_CHECK_EQUAL(log.count("Sub-generator 0 (evtpool) failed to generate the requested event"), 1); +} + +std::map> const& scenarios() +{ + static std::map> const s = { + {"exact_pool", scenarioExactPool}, + {"larger_pool", [](ScenarioEnv const& env) { scenarioLargerPool(env, false); }}, + {"larger_pool_random", [](ScenarioEnv const& env) { scenarioLargerPool(env, true); }}, + {"cocktail_pool", scenarioCocktailPool}, + {"cocktail_groups", scenarioCocktailGroups}, + {"parallel_mode", scenarioParallelMode}, + {"mixed_sequence", scenarioMixedSequence}, + {"uneven_shares", scenarioUnevenShares}, + {"random_shares", scenarioRandomShares}, + {"init_failure", scenarioInitFailure}, + {"init_failure_hepmc", scenarioInitFailureHepMC}, + {"hepmc_share", scenarioHepMCShare}, + {"negative_fraction", scenarioNegativeFraction}, + {"zero_fractions_random", scenarioZeroFractionsRandom}, + {"too_few_events", scenarioTooFewEvents}, + {"mid_job_share", scenarioMidJobShare}, + {"zero_fraction", scenarioZeroFraction}, + {"unused_failure", scenarioUnusedFailure}, + {"failure_forwarded", scenarioFailureForwarded}}; + return s; +} + +// --------------------------------------------------------------------------------------------------------------- +// helpers of the parent process +// --------------------------------------------------------------------------------------------------------------- + +struct ChildResult { + bool exited = false; + int status = -1; // exit code, or the number of the signal which ended the child + std::string output; + + /// the interesting part of the output of the child, for the failure messages + std::string digest() const + { + std::istringstream in(output); + std::vector lines; + for (std::string line; std::getline(in, line);) { + lines.push_back(line); + } + std::string d; + for (size_t i = 0; i < lines.size(); ++i) { + bool interesting = lines[i].find("HYBRIDTEST") != std::string::npos || lines[i].find("error") != std::string::npos || + lines[i].find("fatal") != std::string::npos || lines[i].find("FATAL") != std::string::npos || + lines[i].find("warn") != std::string::npos || lines[i].find("WARN") != std::string::npos || + i + 25 >= lines.size(); + if (interesting) { + d += lines[i] + "\n"; + } + } + return std::string("child ") + (exited ? "exited with code " : "was killed by signal ") + std::to_string(status) + "\n" + d; + } +}; + +/// starts this executable again, with the given scenario +ChildResult runChild(std::string const& scenario, int windowMs = kDefaultWindowMs) +{ + TempDir tmp("hybrid_" + scenario); + auto logFile = (tmp.path / "child.log").string(); + auto exe = std::string(boost::unit_test::framework::master_test_suite().argv[0]); + setenv(kScenarioEnv, scenario.c_str(), 1); + setenv(kDirEnv, (tmp.path / "work").c_str(), 1); + setenv(kWindowEnv, std::to_string(windowMs).c_str(), 1); + // "exec": the shell is replaced by the child, so that a signal ending it is not hidden in an exit code + auto cmd = "exec \"" + exe + "\" --run_test=HybridScenarioWorker > \"" + logFile + "\" 2>&1"; + int rc = std::system(cmd.c_str()); + unsetenv(kScenarioEnv); + unsetenv(kDirEnv); + unsetenv(kWindowEnv); + + ChildResult result; + result.exited = WIFEXITED(rc); + result.status = WIFEXITED(rc) ? WEXITSTATUS(rc) : (WIFSIGNALED(rc) ? WTERMSIG(rc) : -1); + std::ifstream in(logFile); + std::stringstream ss; + ss << in.rdbuf(); + result.output = ss.str(); + return result; +} + +void expectSuccess(std::string const& scenario) +{ + auto result = runChild(scenario); + BOOST_CHECK_MESSAGE(result.exited && result.status == 0, "scenario '" << scenario << "' failed: " << result.digest()); +} + +} // namespace + +// Finding 2: no lookahead after the last event of the job ---------------------------------------------------------- + +BOOST_AUTO_TEST_CASE(Hybrid_ExactPool_NoLookaheadAfterLastEvent) +{ + expectSuccess("exact_pool"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_LargerPool_NoEventBeyondTheJob) +{ + expectSuccess("larger_pool"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_LargerPool_RandomMode_NoEventBeyondTheJob) +{ + expectSuccess("larger_pool_random"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_Cocktail_ExactPool) +{ + expectSuccess("cocktail_pool"); +} + +// the other operation modes ------------------------------------------------------------------------------------ + +BOOST_AUTO_TEST_CASE(Hybrid_Cocktail_Groups) +{ + expectSuccess("cocktail_groups"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_ParallelMode) +{ + expectSuccess("parallel_mode"); +} + +// Findings 4/8: expectations and initialisation of the sub-generators ------------------------------------------------ + +BOOST_AUTO_TEST_CASE(Hybrid_SubGeneratorDoesNotAssumeTheWholeJob) +{ + expectSuccess("mixed_sequence"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_SubGeneratorShare_FromFractions) +{ + auto result = runChild("uneven_shares"); + BOOST_CHECK_MESSAGE(result.exited && result.status == kFatalExitCode, "expected the fatal exit of the child: " << result.digest()); + BOOST_CHECK_MESSAGE(result.output.find("HYBRIDTEST shares=7,3, kine=7") != std::string::npos, + "the shares computed from the fractions are wrong: " << result.digest()); + BOOST_CHECK_MESSAGE(result.output.find("This job will request 7 events, but the input (1 file(s), 6 events in the first one) holds only about 6") != std::string::npos, + "the pool did not warn about its share: " << result.digest()); + BOOST_CHECK_MESSAGE(result.output.find("ran out of events after 6 event(s) from 1 input file(s) (7 were requested)") != std::string::npos, + "the failure does not report the share of the pool: " << result.digest()); +} + +BOOST_AUTO_TEST_CASE(Hybrid_SubGeneratorShare_Randomized) +{ + expectSuccess("random_shares"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_SubGeneratorInitFailureIsFatal) +{ + expectSuccess("init_failure"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_SubGeneratorInitResultIsChecked) +{ + expectSuccess("init_failure_hepmc"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_HepMCSubGenerator_KnowsItsShare) +{ + expectSuccess("hepmc_share"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_NegativeFraction_IsRejected) +{ + expectSuccess("negative_fraction"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_ZeroFractionsRandom_IsRejected) +{ + expectSuccess("zero_fractions_random"); +} + +// Finding 3: a pool which is too small remains a fatal error --------------------------------------------------------- + +BOOST_AUTO_TEST_CASE(Hybrid_PoolTooSmall_IsFatal) +{ + auto result = runChild("too_few_events"); + BOOST_CHECK_MESSAGE(result.exited && result.status == kFatalExitCode, "expected the fatal exit of the child: " << result.digest()); + BOOST_CHECK_MESSAGE(result.output.find("ran out of events after 9 event(s) from 3 input file(s)") != std::string::npos, + "the reason of the failure is not reported: " << result.digest()); + // the only generator of the hybrid is asked for all the events of the job, and knows it + BOOST_CHECK_MESSAGE(result.output.find("ran out of events after 9 event(s) from 3 input file(s) (10 were requested)") != std::string::npos, + "the number of requested events is not reported: " << result.digest()); +} + +// Follow-up of finding 2: request budget, failures in the worker threads ---------------------------------------------- + +BOOST_AUTO_TEST_CASE(Hybrid_PoolUsedUpMidJob_NotAskedBeyondItsShare) +{ + expectSuccess("mid_job_share"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_ZeroFraction_NeverAsked) +{ + expectSuccess("zero_fraction"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_UnconsumedLookaheadFailure_IsHarmless) +{ + expectSuccess("unused_failure"); +} + +BOOST_AUTO_TEST_CASE(Hybrid_WorkerFailure_ReachesMainThread) +{ + expectSuccess("failure_forwarded"); +} + +// the child process -------------------------------------------------------------------------------------------- + +/// Runs the scenario given by the environment; started by the test cases above, a no-op otherwise. +BOOST_AUTO_TEST_CASE(HybridScenarioWorker) +{ + const char* scenario = std::getenv(kScenarioEnv); + if (!scenario) { + return; + } + // no core dumps from the scenarios which fail, and no hanging forever if something goes wrong; a watchdog + // thread rather than alarm(): Boost.Test handles SIGALRM itself, and a hung hybrid then blocks the exit + rlimit noCore{0, 0}; + setrlimit(RLIMIT_CORE, &noCore); + std::thread([] { + std::this_thread::sleep_for(std::chrono::seconds(kWatchdogSeconds)); + std::cout << "HYBRIDTEST watchdog: the scenario did not end within " << kWatchdogSeconds << " s" << std::endl; + std::_Exit(kWatchdogExitCode); + }).detach(); + + auto it = scenarios().find(scenario); + BOOST_REQUIRE_MESSAGE(it != scenarios().end(), "unknown scenario " << scenario); + ScenarioEnv env; + env.dir = std::getenv(kDirEnv) ? std::getenv(kDirEnv) : "hybrid_scenario"; + env.windowMs = std::getenv(kWindowEnv) ? std::atoi(std::getenv(kWindowEnv)) : kDefaultWindowMs; + fs::create_directories(env.dir); + it->second(env); +} diff --git a/run/SimExamples/EventPool_Chaining/README.md b/run/SimExamples/EventPool_Chaining/README.md new file mode 100644 index 0000000000000..290d9d4ba018c --- /dev/null +++ b/run/SimExamples/EventPool_Chaining/README.md @@ -0,0 +1,149 @@ + + +# Chaining of event pools + +The `evtpool` generator goes through the **whole pool**, one file at a time: it starts +with the first one and, once its events are exhausted, moves on to the next. The files +are opened strictly one at a time and only when they are actually needed — a pool of a +thousand files costs exactly one open file handle, and a job that never gets past the +first file never touches the others. This is what is needed on hyperloop, where many +events have to be served from a pool made of many small files. + +# Configuration + +| key | default | meaning | +| --- | --- | --- | +| `GeneratorEventPool.eventPoolPath` | `""` | pool directory, a text file with a list of files, or a single `evtpool.root` | +| `GeneratorEventPool.randomize` | `true` | serve the events of each file in random order (a permutation, every event exactly once) | +| `GeneratorEventPool.roundRobin` | `false` | start over with the first file once the last one is exhausted | +| `GeneratorEventPool.rngseed` | `0` | seed used both for the order in which files are visited and for the event randomization | + +The order in which the pool files are visited is shuffled per job, so that different jobs +of the same production do not all read the same files in the same order. Using a fixed +`rngseed` makes that order reproducible. No file is opened at initialisation time apart +from the first one — the shuffle only picks *names*. + +Order in which events are served — it is fixed for a file at the moment that file is +opened, and every one of its events is used **exactly once** (different behaviour than the past): + +* `randomize=false`: entry 0, 1, 2, … of the first file, then entry 0, 1, 2, … of the + second file, and so on; +* `randomize=true` (the default): a random permutation of the entries of the first file, + then a random permutation of the entries of the second one, and so on. + +Once the last file has been used up the job **fails with a fatal error** unless +`roundRobin=true` is set. Making sure that the pool holds enough events for the +requested `-n` is the responsibility of the user: + +``` +[FATAL] GeneratorFromO2Kine: ran out of events after 6 event(s) from 3 input file(s) + (9 were requested). Provide more input files/events or allow reusing them via roundRobin +``` + +With `roundRobin=true` the generator starts over from the first file; in randomized mode +a fresh permutation is drawn on every pass. + +Example: + +```bash +o2-sim -n 100 -g evtpool --configKeyValues "GeneratorEventPool.eventPoolPath=/path/to/pool" +``` + +The same works for AliEn pools: + +```bash +o2-sim -n 100 -g evtpool \ + --configKeyValues "GeneratorEventPool.eventPoolPath=alien:///alice/cern.ch/user/.../evtpool_dir" +``` + +and inside the hybrid generator JSON configuration: + +```json +{ + "name": "evtpool", + "config": { + "eventPoolPath": "/path/to/pool", + "skipNonTrackable": true, + "roundRobin": false, + "randomize": false, + "rngseed": 0, + "randomphi": false + } +} +``` + +## Use from the DPL event generator (hyperloop) + +The very same configuration works with `o2-sim-dpl-eventgen`, which is the entry point +used on hyperloop: + +```bash +o2-sim-dpl-eventgen -b --nEvents 1000 --generator evtpool --vertexMode kNoVertex \ + --configKeyValues "GeneratorEventPool.eventPoolPath=/path/to/pool" |\ + o2-sim-mctracks-to-aod -b |\ + o2-analysis-mctracks-to-aod-simple-task -b +``` + +`rundpl.sh` in this folder runs that pipeline for the pool created by `run.sh`. + +Note that the seed given to the driver (`o2-sim --seed` / `o2-sim-dpl-eventgen --seed`) +governs both the order in which the pool files are visited and the event order inside them, +as long as `GeneratorEventPool.rngseed` is left at its default of 0. Setting `rngseed` +explicitly overrides the driver seed for the event pool. + +Example of a workflow reading from an event pool: + +```bash +${O2DPG_ROOT}/MC/bin/o2dpg_sim_workflow.py -eCM 900 -col pp -gen evtpool -tf 1 -ns 7 \ + -e TGeant4 -j 4 -interactionRate 50000 -run 300000 -seed 12345 \ + -confKey "GeneratorEventPool.eventPoolPath=/path/to/pool" +${O2DPG_ROOT}/MC/bin/o2dpg_workflow_runner.py -f workflow.json -tt aod +``` + +## Reading several plain kinematics files (`extkinO2`) + +The same machinery is available for the ordinary `extkinO2` generator by passing a +comma-separated list of files, which are again read one after the other: + +```bash +o2-sim -n 100 -g extkinO2 --extKinFile "kine1.root,kine2.root,kine3.root" +# or +o2-sim -n 100 -g extkinO2 --configKeyValues "GeneratorFromO2Kine.fileName=kine1.root,kine2.root" +``` + +## Running out of events, and use inside `GeneratorHybrid` + +* Every event of the pool is served exactly once per pass. Before the chaining was introduced, + `evtpool` sampled the events *with replacement* and could never run out. Now, asking for more + events than the pool holds is a **fatal** error unless `roundRobin=true` is set: provide enough + input, or enable `roundRobin` to start over once everything has been used. +* Inside a `GeneratorHybrid` a pool only serves its share of the events of the job. The hybrid + computes this share from the fractions (exact in sequential and cocktail mode, the mean with + `randomize`, unknown in `parallel` mode) and passes it to the pool, which uses it for the + "input too small" warning at initialisation and in the fatal message. +* With fixed fractions (sequential or cocktail mode) the hybrid never asks a sub-generator for + more events than its share, so a pool holding exactly its share is not read beyond its end, + wherever in the job it is used up. In random and `parallel` mode no event is scheduled after the + last event of the job, and a scheduled event which cannot be generated is only an error if the + job actually uses it. +* An input which cannot be used at all (a pool path without any `evtpool.root`, or a list of + kinematics files none of which can be read) stops the job at initialisation, with or without + `GeneratorHybrid`. Single unreadable files of a list are skipped when they are reached. +* A sub-generator of a hybrid which fails to initialise stops the job at initialisation; one which + fails while generating an event that the job needs stops the job, also when the hybrid runs + several worker threads. + +# Provenance of the events + +Every generated event stores the file it was read from and the entry inside that file in +its MC event header: + +* `forwarding-generator_inputFile` +* `forwarding-generator_inputEventNumber` (entry within that file) + +# Files description + +- **run.sh** → creates a small event pool and reads it back +- **rundpl.sh** → the same through `o2-sim-dpl-eventgen` (the hyperloop path) diff --git a/run/SimExamples/EventPool_Chaining/run.sh b/run/SimExamples/EventPool_Chaining/run.sh new file mode 100755 index 0000000000000..dc7efc8a771b1 --- /dev/null +++ b/run/SimExamples/EventPool_Chaining/run.sh @@ -0,0 +1,76 @@ +#!/usr/bin/env bash +# +# Example showing how a job goes through an event pool made of several files: one +# file is read at a time and the next one is opened only once the events of the +# current one are exhausted. +# +# Stage 1: create a small event pool made of NPOOLFILES files +# Stage 2: read the pool back in its different modes +# +# Note: the example runs with '--noGeant' and a minimal geometry ('-m PIPE'), +# since only the event generation part is of interest here. +# +set -x +set -e + +[ ! "${O2_ROOT}" ] && echo "Error: This needs O2 loaded" && exit 1 + +NPOOLFILES=3 # number of files the pool consists of +NEVENTS_PER_FILE=3 # events per pool file +NEVENTS=9 # events to be read back from the whole pool + +POOLDIR=${PWD}/eventpool +COMMON="-j 4 --noGeant -m PIPE --vertexMode kNoVertex" + +# --------------------------------------------------------------------------- +# Stage 1: produce the event pool +# --------------------------------------------------------------------------- +# An event pool is a set of kinematics files that are all called 'evtpool.root' +# and that live in separate sub-directories of a common pool directory. +rm -rf ${POOLDIR} +for i in $(seq 0 $((NPOOLFILES - 1))); do + o2-sim ${COMMON} -n ${NEVENTS_PER_FILE} -g pythia8pp -o poolgen_${i} --seed $((i + 1)) + mkdir -p ${POOLDIR}/00${i} + mv poolgen_${i}_Kine.root ${POOLDIR}/00${i}/evtpool.root +done + +# --------------------------------------------------------------------------- +# Stage 2a: sequential roll-over through the whole pool +# --------------------------------------------------------------------------- +# The generator moves on to the next pool file once the events of the current one +# are used up, so that NPOOLFILES * NEVENTS_PER_FILE events are available in total. +# Only one file is open at a time. +o2-sim ${COMMON} -n ${NEVENTS} -g evtpool -o rollover \ + --configKeyValues "GeneratorEventPool.eventPoolPath=${POOLDIR};GeneratorEventPool.randomize=false;GeneratorEventPool.rngseed=1" + +# --------------------------------------------------------------------------- +# Stage 2b: start over with the first file once the pool is exhausted +# --------------------------------------------------------------------------- +o2-sim ${COMMON} -n $((NEVENTS + NEVENTS_PER_FILE)) -g evtpool -o rollover_roundrobin \ + --configKeyValues "GeneratorEventPool.eventPoolPath=${POOLDIR};GeneratorEventPool.roundRobin=true;GeneratorEventPool.randomize=false;GeneratorEventPool.rngseed=1" + +# --------------------------------------------------------------------------- +# Stage 2c: randomized access (the event pool default) across the files +# --------------------------------------------------------------------------- +# The entries of each file are served in a random permutation, so every event of +# the pool is still used exactly once. +o2-sim ${COMMON} -n ${NEVENTS} -g evtpool -o rollover_random \ + --configKeyValues "GeneratorEventPool.eventPoolPath=${POOLDIR};GeneratorEventPool.rngseed=1" + +# --------------------------------------------------------------------------- +# Stage 2d: asking for more events than the pool holds must fail +# --------------------------------------------------------------------------- +# It is up to the user to provide enough events; running out of them is fatal. +if o2-sim ${COMMON} -n $((NEVENTS + 1)) -g evtpool -o toofew \ + --configKeyValues "GeneratorEventPool.eventPoolPath=${POOLDIR};GeneratorEventPool.rngseed=1"; then + echo "ERROR: the simulation should have failed with 'ran out of events'" && exit 1 +else + echo "OK: the simulation failed as expected (not enough events in the pool)" +fi + +# --------------------------------------------------------------------------- +# Stage 2e: the same mechanism is available for plain 'extkinO2' by giving a +# comma-separated list of files +# --------------------------------------------------------------------------- +FILELIST=$(ls ${POOLDIR}/*/evtpool.root | paste -sd,) +o2-sim ${COMMON} -n ${NEVENTS} -g extkinO2 -o extkin_rollover --extKinFile "${FILELIST}" diff --git a/run/SimExamples/EventPool_Chaining/rundpl.sh b/run/SimExamples/EventPool_Chaining/rundpl.sh new file mode 100755 index 0000000000000..b7e03ea091438 --- /dev/null +++ b/run/SimExamples/EventPool_Chaining/rundpl.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# +# Same event pool roll-over as run.sh, but through the DPL event generator +# i.e. the path used on hyperloop: +# +# o2-sim-dpl-eventgen | o2-sim-mctracks-to-aod | +# +# Requires the event pool created by run.sh (or any other pool directory given +# via the POOLDIR variable). +# +set -x +set -e +set -o pipefail + +[ ! "${O2_ROOT}" ] && echo "Error: This needs O2 loaded" && exit 1 + +POOLDIR=${POOLDIR:-${PWD}/eventpool} +[ ! -d "${POOLDIR}" ] && echo "Error: no event pool at ${POOLDIR}; run run.sh first" && exit 2 + +NEVENTS=9 +COMMON="-b --aggregate-timeframe 10 --vertexMode kNoVertex --seed 12345" + +# --------------------------------------------------------------------------- +# a) the whole pool as one event stream +# --------------------------------------------------------------------------- +o2-sim-dpl-eventgen ${COMMON} --nEvents ${NEVENTS} --generator evtpool \ + --configKeyValues "GeneratorEventPool.eventPoolPath=${POOLDIR}" |\ + o2-sim-mctracks-to-aod -b |\ + o2-analysis-mctracks-to-aod-simple-task -b + +# --------------------------------------------------------------------------- +# b) reuse the pool once it is exhausted (2 full passes here) +# --------------------------------------------------------------------------- +o2-sim-dpl-eventgen ${COMMON} --nEvents $((2 * NEVENTS)) --generator evtpool \ + --configKeyValues "GeneratorEventPool.eventPoolPath=${POOLDIR};GeneratorEventPool.roundRobin=true" |\ + o2-sim-mctracks-to-aod -b |\ + o2-analysis-mctracks-to-aod-simple-task -b + +# --------------------------------------------------------------------------- +# c) plain 'extkinO2' with a comma-separated list of files +# --------------------------------------------------------------------------- +FILELIST=$(ls ${POOLDIR}/*/evtpool.root | paste -sd,) +o2-sim-dpl-eventgen ${COMMON} --nEvents ${NEVENTS} --generator extkinO2 \ + --configKeyValues "GeneratorFromO2Kine.fileName=${FILELIST}" |\ + o2-sim-mctracks-to-aod -b |\ + o2-analysis-mctracks-to-aod-simple-task -b + +# --------------------------------------------------------------------------- +# d) asking for more events than the pool holds must fail, not hang +# --------------------------------------------------------------------------- +if o2-sim-dpl-eventgen ${COMMON} --nEvents $((NEVENTS + 1)) --generator evtpool \ + --configKeyValues "GeneratorEventPool.eventPoolPath=${POOLDIR}" |\ + o2-sim-mctracks-to-aod -b |\ + o2-analysis-mctracks-to-aod-simple-task -b; then + echo "ERROR: the workflow should have failed with 'ran out of events'" && exit 1 +else + echo "OK: the workflow failed as expected (not enough events in the pool)" +fi diff --git a/run/SimExamples/README.md b/run/SimExamples/README.md index c276c05b650f2..4030b028c938a 100644 --- a/run/SimExamples/README.md +++ b/run/SimExamples/README.md @@ -24,6 +24,7 @@ * \subpage refrunSimExamplesPythia * \subpage refrunSimExamplesForceDecay_Lambda_Neutron_Dalitz * \subpage refrunSimExamplesJustPrimaryKinematics +* \subpage refrunSimExamplesEventPool_Chaining * \subpage refrunSimExamplesMergeEventPools * \subpage refrunSimExamplesSelective_Transport * \subpage refrunSimExamplesSelective_Transport_pi0