Apply correct TBB shutdown mechanism

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);

Generators/src/GeneratorHybrid.cxx

@@ -10,6 +10,7 @@
 // or submit itself to any jurisdiction.
 
 #include "Generators/GeneratorHybrid.h"
+#include <chrono>
 #include <fairlogger/Logger.h>
 #include <algorithm>
 #include <tbb/concurrent_queue.h>
@@ -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::milliseconds>(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++;