diff --git a/Generators/include/Generators/GeneratorHybrid.h b/Generators/include/Generators/GeneratorHybrid.h index a47ef0b7cce98..061eed088e5d3 100644 --- a/Generators/include/Generators/GeneratorHybrid.h +++ b/Generators/include/Generators/GeneratorHybrid.h @@ -64,6 +64,7 @@ class GeneratorHybrid : public Generator Bool_t generateEvent() override; Bool_t importParticles() override; void updateHeader(o2::dataformats::MCEventHeader* eventHeader) override; + void stop() override; // end-of-stream hook that joins the TBB task pool Bool_t parseJSON(const std::string& path); Bool_t confSetter(const auto& gen); diff --git a/Generators/src/GeneratorHybrid.cxx b/Generators/src/GeneratorHybrid.cxx index 29ea6e6737a8c..1afb9446cbe05 100644 --- a/Generators/src/GeneratorHybrid.cxx +++ b/Generators/src/GeneratorHybrid.cxx @@ -10,6 +10,7 @@ // or submit itself to any jurisdiction. #include "Generators/GeneratorHybrid.h" +#include #include #include #include @@ -191,11 +192,38 @@ GeneratorHybrid::GeneratorHybrid(const std::string& inputgens) GeneratorHybrid::~GeneratorHybrid() { LOG(info) << "Destructor of generator hybrid called"; + // stop() is the designated shutdown hook (called via GeneratorService::stopGenerators), + // normally running end-of-stream, in which case this is a no-op. It is not reached on every exit path, e.g. when the + // driver quits before the event budget or the TTL is up, so it is repeated here. + stop(); +} + +void GeneratorHybrid::stop() +{ mStopFlag = true; + // The worker lambdas reference mInputTaskQueue, mResultQueue and mTaskArena, which are + // destroyed together with this object, so they must not be left running. + // We wait for the thread to finish, which will happen once the worker lambdas notice mStopFlag. + if (mTBBTaskPoolRunner.joinable()) { + auto t0 = std::chrono::steady_clock::now(); + mTBBTaskPoolRunner.join(); + auto ms = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + LOG(info) << "HybridGen: stop() joined TBB task pool after " << ms << " ms"; + } + // Forward the hook to the sub-generators + for (auto& gen : gens) { + if (gen) { + gen->stop(); + } + } } Bool_t GeneratorHybrid::Init() { + // protection against additional Init() which would assign over a joinable mTBBTaskPoolRunner, which calls std::terminate + if (mIsInitialized) { + return true; + } // init all sub-gens int count = 0; for (auto& gen : mGens) { @@ -320,7 +348,7 @@ Bool_t GeneratorHybrid::Init() // } } bool isTriggered = false; - while (!isTriggered) { + while (!isTriggered && !mStopFlag) { generator->clearParticles(); generator->generateEvent(); generator->importParticles(); @@ -355,8 +383,8 @@ Bool_t GeneratorHybrid::Init() // let the TBB task system run in it's own thread mTBBTaskPoolRunner = std::thread([this, worker_function]() { mTaskArena.execute([&]() { tbb::parallel_for(0, mTaskArena.max_concurrency(), [&](int) { worker_function(); }); }); }); - mTBBTaskPoolRunner.detach(); // detaching because we don't like to wait on the thread to finish - // some of the generators might still be generating when we are done + // NOTE: thread deliberately NOT detached. We never wait for it during the run, but the + // pool must be joined in the destructor before the queues and the arena are destroyed. // let's also push initial generation tasks for each event generator for (size_t genindex = 0; genindex < gens.size(); ++genindex) { @@ -456,6 +484,9 @@ bool GeneratorHybrid::importParticles() // this is important when using event pools where the full header information is forwarded from the generator // otherwise some events might have mixed header information from different generators mMCEventHeader.Reset(); + // Once the event budget is exhausted there is no further event to prepare, and re-queueing a + // sub-generator would start an event nobody consumes and that stop() then has to wait for + const bool queueNextEvent = getTotalNEvents() == 0 || mEventCounter + 1 < getTotalNEvents(); if (mCocktailMode) { // in cocktail mode we need to merge the particles from the different generators for (auto subIndex : subGenIndex) { @@ -486,8 +517,10 @@ bool GeneratorHybrid::importParticles() gens[subIndex]->updateHeader(&mMCEventHeader); mHeaderGeneratorIndex = subIndex; // store index of generator updating the header } - mInputTaskQueue.push(subIndex); - mTasksStarted++; + if (queueNextEvent) { + mInputTaskQueue.push(subIndex); + mTasksStarted++; + } } } else { if (!mIsHyperloop) { @@ -510,8 +543,10 @@ 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 (queueNextEvent) { + mInputTaskQueue.push(genIndex); + mTasksStarted++; + } } mseqCounter++;