Skip to content

Commit 58f90b8

Browse files
authored
Join the hybrid generator's TBB pool on shutdown (#16008)
This fixes a segfault at the end of dpl-eventgen with the hybrid generator, seen in the Hyperloop tests. - The TBB pool thread was detached and could outlive the generator. It is now joined in the new stop() hook and in the destructor. - stop() is forwarded to the sub-generators, so external generator processes are reaped. - No new event is queued once the event budget is reached. - A second Init() is now a no-op. Related to #15636
1 parent 6d434fa commit 58f90b8

2 files changed

Lines changed: 43 additions & 7 deletions

File tree

‎Generators/include/Generators/GeneratorHybrid.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ class GeneratorHybrid : public Generator
6464
Bool_t generateEvent() override;
6565
Bool_t importParticles() override;
6666
void updateHeader(o2::dataformats::MCEventHeader* eventHeader) override;
67+
void stop() override; // end-of-stream hook that joins the TBB task pool
6768

6869
Bool_t parseJSON(const std::string& path);
6970
Bool_t confSetter(const auto& gen);

‎Generators/src/GeneratorHybrid.cxx‎

Lines changed: 42 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
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)
191192
GeneratorHybrid::~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

197221
Bool_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

Comments
 (0)