1010// or submit itself to any jurisdiction.
1111
1212#include " Generators/GeneratorHybrid.h"
13+ #include < chrono>
1314#include < fairlogger/Logger.h>
1415#include < algorithm>
1516#include < tbb/concurrent_queue.h>
@@ -191,11 +192,38 @@ GeneratorHybrid::GeneratorHybrid(const std::string& inputgens)
191192GeneratorHybrid::~GeneratorHybrid ()
192193{
193194 LOG (info) << " Destructor of generator hybrid called" ;
195+ // stop() is the designated shutdown hook (called via GeneratorService::stopGenerators),
196+ // 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
197+ // driver quits before the event budget or the TTL is up, so it is repeated here.
198+ stop ();
199+ }
200+
201+ void GeneratorHybrid::stop ()
202+ {
194203 mStopFlag = true ;
204+ // The worker lambdas reference mInputTaskQueue, mResultQueue and mTaskArena, which are
205+ // destroyed together with this object, so they must not be left running.
206+ // We wait for the thread to finish, which will happen once the worker lambdas notice mStopFlag.
207+ if (mTBBTaskPoolRunner .joinable ()) {
208+ auto t0 = std::chrono::steady_clock::now ();
209+ mTBBTaskPoolRunner .join ();
210+ auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now () - t0).count ();
211+ LOG (info) << " HybridGen: stop() joined TBB task pool after " << ms << " ms" ;
212+ }
213+ // Forward the hook to the sub-generators
214+ for (auto & gen : gens) {
215+ if (gen) {
216+ gen->stop ();
217+ }
218+ }
195219}
196220
197221Bool_t GeneratorHybrid::Init ()
198222{
223+ // protection against additional Init() which would assign over a joinable mTBBTaskPoolRunner, which calls std::terminate
224+ if (mIsInitialized ) {
225+ return true ;
226+ }
199227 // init all sub-gens
200228 int count = 0 ;
201229 for (auto & gen : mGens ) {
@@ -320,7 +348,7 @@ Bool_t GeneratorHybrid::Init()
320348 // }
321349 }
322350 bool isTriggered = false ;
323- while (!isTriggered) {
351+ while (!isTriggered && ! mStopFlag ) {
324352 generator->clearParticles ();
325353 generator->generateEvent ();
326354 generator->importParticles ();
@@ -355,8 +383,8 @@ Bool_t GeneratorHybrid::Init()
355383
356384 // let the TBB task system run in it's own thread
357385 mTBBTaskPoolRunner = std::thread ([this , worker_function]() { mTaskArena .execute ([&]() { tbb::parallel_for (0 , mTaskArena .max_concurrency (), [&](int ) { worker_function (); }); }); });
358- mTBBTaskPoolRunner . detach (); // detaching because we don't like to wait on the thread to finish
359- // some of the generators might still be generating when we are done
386+ // NOTE: thread deliberately NOT detached. We never wait for it during the run, but the
387+ // pool must be joined in the destructor before the queues and the arena are destroyed.
360388
361389 // let's also push initial generation tasks for each event generator
362390 for (size_t genindex = 0 ; genindex < gens.size (); ++genindex) {
@@ -456,6 +484,9 @@ bool GeneratorHybrid::importParticles()
456484 // this is important when using event pools where the full header information is forwarded from the generator
457485 // otherwise some events might have mixed header information from different generators
458486 mMCEventHeader .Reset ();
487+ // Once the event budget is exhausted there is no further event to prepare, and re-queueing a
488+ // sub-generator would start an event nobody consumes and that stop() then has to wait for
489+ const bool queueNextEvent = getTotalNEvents () == 0 || mEventCounter + 1 < getTotalNEvents ();
459490 if (mCocktailMode ) {
460491 // in cocktail mode we need to merge the particles from the different generators
461492 for (auto subIndex : subGenIndex) {
@@ -486,8 +517,10 @@ bool GeneratorHybrid::importParticles()
486517 gens[subIndex]->updateHeader (&mMCEventHeader );
487518 mHeaderGeneratorIndex = subIndex; // store index of generator updating the header
488519 }
489- mInputTaskQueue .push (subIndex);
490- mTasksStarted ++;
520+ if (queueNextEvent) {
521+ mInputTaskQueue .push (subIndex);
522+ mTasksStarted ++;
523+ }
491524 }
492525 } else {
493526 if (!mIsHyperloop ) {
@@ -510,8 +543,10 @@ bool GeneratorHybrid::importParticles()
510543 // fetch the event Header information from the underlying generator
511544 gens[genIndex]->updateHeader (&mMCEventHeader );
512545 mHeaderGeneratorIndex = genIndex; // store index of generator updating the header
513- mInputTaskQueue .push (genIndex);
514- mTasksStarted ++;
546+ if (queueNextEvent) {
547+ mInputTaskQueue .push (genIndex);
548+ mTasksStarted ++;
549+ }
515550 }
516551
517552 mseqCounter++;
0 commit comments