From f595dcc10155e2897f908bf2eefa8d649259701f Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 24 Sep 2026 20:52:47 +0200 Subject: [PATCH 1/7] Fix the merger exit-status check in o2-sim This fixes a problem in how the o2-sim driver judges the hit merger's exit code. - The condition `!= 0 || != 128` is always true, so every normal merger exit was reported as an error. - It now reads `!= 0 && != 128`. Co-Authored-By: Claude Opus 5.5 --- run/o2sim_parallel.cxx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/run/o2sim_parallel.cxx b/run/o2sim_parallel.cxx index 8a92a5f251cb0..0d5ff361141a8 100644 --- a/run/o2sim_parallel.cxx +++ b/run/o2sim_parallel.cxx @@ -763,7 +763,7 @@ int main(int argc, char* argv[]) // Handle mergerpid status separately if (cpid == mergerpid) { if (WIFEXITED(status)) { - if (WEXITSTATUS(status) != 0 || WEXITSTATUS(status) != 128) { + if (WEXITSTATUS(status) != 0 && WEXITSTATUS(status) != 128) { LOG(error) << "Merger process exited with abnormal exit status " << WEXITSTATUS(status); errored = true; } From 7939bca2ecf000afa02c3a8224c30811f1b2b54d Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 24 Sep 2026 20:52:52 +0200 Subject: [PATCH 2/7] Do not cache external-kinematics generators in the primary server This fixes the generator cache of the o2-sim primary server so that it skips external kinematics, as its comment intends. - The condition `!= "extkin" || != "extkinO2"` is always true, so extkin generators were cached too. - A service-mode reconfiguration with a new kinematics file then reused the old generator and file. - The condition now uses `&&`. Co-Authored-By: Claude Opus 5.5 --- run/O2PrimaryServerDevice.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/run/O2PrimaryServerDevice.h b/run/O2PrimaryServerDevice.h index b8703ffcddb28..77a3bf9f269e1 100644 --- a/run/O2PrimaryServerDevice.h +++ b/run/O2PrimaryServerDevice.h @@ -108,7 +108,7 @@ class O2PrimaryServerDevice final : public fair::mq::Device // Not using cached instances for external kinematics since these might change input filenames etc. // and are in any case quickly setup. mPrimGen = nullptr; - if (conf.getGenerator().compare("extkin") != 0 || conf.getGenerator().compare("extkinO2") != 0) { + if (conf.getGenerator().compare("extkin") != 0 && conf.getGenerator().compare("extkinO2") != 0) { auto iter = mPrimGeneratorCache.find(conf.getGenerator()); if (iter != mPrimGeneratorCache.end()) { mPrimGen = iter->second.get(); From 0d3516030db8c468907245d14c8672d5050e53c8 Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 24 Sep 2026 20:52:58 +0200 Subject: [PATCH 3/7] Skip malformed info requests in the primary server This fixes the info thread of the o2-sim primary server for requests of unexpected size. - After the error reply the thread still read the request and could send a second reply on the REP socket. - It now continues to the next request after the error reply. Co-Authored-By: Claude Opus 5.5 --- run/O2PrimaryServerDevice.h | 1 + 1 file changed, 1 insertion(+) diff --git a/run/O2PrimaryServerDevice.h b/run/O2PrimaryServerDevice.h index 77a3bf9f269e1..d5608078593d4 100644 --- a/run/O2PrimaryServerDevice.h +++ b/run/O2PrimaryServerDevice.h @@ -273,6 +273,7 @@ class O2PrimaryServerDevice final : public fair::mq::Device if (request->GetSize() != sizeof(request_payload)) { LOG(error) << "Obtained request with unexpected payload size"; sendErrorReply(channel); // ALWAYS reply + continue; } memcpy(&request_payload, request->GetData(), sizeof(request_payload)); From c372745f38c0140d2686091385ebf63e27e55e99 Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 24 Sep 2026 20:53:24 +0200 Subject: [PATCH 4/7] Fix the event-flush loop of the o2-sim hit merger This fixes how the hit merger advances to the next event while flushing. - The skip paths advanced the event counter but then carried on with the current event. - With --noemptyevents, an event without hits was written when the next event was already complete, and that next event was then skipped. - An event without buffered info dereferenced the end iterator. - The loop now iterates over flushable event IDs and every skip path is a plain continue. Co-Authored-By: Claude Opus 5.5 --- run/O2HitMerger.h | 35 ++++++++++------------------------- 1 file changed, 10 insertions(+), 25 deletions(-) diff --git a/run/O2HitMerger.h b/run/O2HitMerger.h index 15f58c6dba351..ea81e16f7206f 100644 --- a/run/O2HitMerger.h +++ b/run/O2HitMerger.h @@ -654,34 +654,24 @@ class O2HitMerger : public fair::mq::Device // The method can be called asynchronously to data collection bool mergeAndFlushData() { - auto checkIfNextFlushable = [this]() -> bool { - mNextFlushID++; - return mFlushableEvents.find(mNextFlushID) != mFlushableEvents.end() && mFlushableEvents[mNextFlushID] == true; + auto isFlushable = [this](int eventID) { + auto iter = mFlushableEvents.find(eventID); + return iter != mFlushableEvents.end() && iter->second; }; LOG(info) << "Launching merge kernel "; - bool canflush = mFlushableEvents.find(mNextFlushID) != mFlushableEvents.end() && mFlushableEvents[mNextFlushID] == true; - if (!canflush) { + if (!isFlushable(mNextFlushID)) { return false; } - while (canflush == true) { + for (; isFlushable(mNextFlushID); ++mNextFlushID) { auto flusheventID = mNextFlushID; LOG(info) << "Merge and flush event " << flusheventID; auto iter = mSubEventInfoBuffer.find(flusheventID); - if (iter == mSubEventInfoBuffer.end()) { - LOG(error) << "No info/data found for event " << flusheventID; - if (!checkIfNextFlushable()) { - return false; - } - } - - auto& subEventInfoList = (*iter).second; - if (subEventInfoList.size() == 0 || mNExpectedEvents == 0) { + if (iter == mSubEventInfoBuffer.end() || iter->second.size() == 0 || mNExpectedEvents == 0) { LOG(error) << "No data entries found for event " << flusheventID; - if (!checkIfNextFlushable()) { - return false; - } + continue; } + auto& subEventInfoList = iter->second; TStopwatch timer; timer.Start(); @@ -716,9 +706,7 @@ class O2HitMerger : public fair::mq::Device if (eventheader && eventheader->getMCEventStats().getNHits() == 0) { LOG(info) << " Taking out event " << flusheventID << " due to no hits "; cleanEvent(flusheventID); - if (!checkIfNextFlushable()) { - return true; - } + continue; } } @@ -834,10 +822,7 @@ class O2HitMerger : public fair::mq::Device cleanEvent(flusheventID); LOG(info) << "Merge/flush for event " << flusheventID << " took " << timer.RealTime(); - if (!checkIfNextFlushable()) { - break; - } - } // end while + } if (mWriteToDisc && mOutFile) { LOG(info) << "Writing TTrees"; mOutFile->Write("", TObject::kOverwrite); From 72dab53d4d7721d234f4f5b24b317739e9651549 Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 24 Sep 2026 20:53:52 +0200 Subject: [PATCH 5/7] Release per-event buffers in the o2-sim hit merger This fixes a memory leak in the hit merger and makes its merge flag thread-safe. - cleanEvent was empty, so the decoded SubEventInfo objects of every event were never freed. - The MC tracks and track references of events dropped by --noemptyevents were never freed either. - cleanEvent now owns the release of all three buffers; the merge functions no longer delete. - Buffer entries are emptied rather than erased, since erasing from a tbb::concurrent_unordered_map is not safe while the receiving thread inserts. - handleSimData keeps the event id and event count as values, since a merge thread may free the event info. - mergingInProgress is shared by two threads and is now std::atomic. Co-Authored-By: Claude Opus 5.5 --- run/O2HitMerger.h | 40 +++++++++++++++++++++++----------------- 1 file changed, 23 insertions(+), 17 deletions(-) diff --git a/run/O2HitMerger.h b/run/O2HitMerger.h index ea81e16f7206f..5814139c27cc1 100644 --- a/run/O2HitMerger.h +++ b/run/O2HitMerger.h @@ -67,6 +67,7 @@ #include #include #include +#include #include #include @@ -399,6 +400,9 @@ class O2HitMerger : public fair::mq::Device int index = 0; auto infoptr = o2::base::decodeTMessage(data, index++); o2::data::SubEventInfo& info = *infoptr; + // once a merge thread runs, the buffered info of a complete event may be freed at any time + const auto eventID = info.eventID; + const auto maxEvents = info.maxEvents; auto accum = insertAdd(mPartsCheckSum, info.eventID, (uint32_t)info.part); LOG(info) << "SIMDATA channel got " << data.Size() << " parts for event " << info.eventID << " part " << info.part << " out of " << info.nparts; @@ -423,19 +427,19 @@ class O2HitMerger : public fair::mq::Device mMergerIOThread.join(); } // start hit merging and flushing in a separate thread in order not to block - mMergerIOThread = std::thread([info, this]() { mergingInProgress = true; mergeAndFlushData(); mergingInProgress = false; }); + mMergerIOThread = std::thread([this]() { mergingInProgress = true; mergeAndFlushData(); mergingInProgress = false; }); } - mEventChecksum += info.eventID; + mEventChecksum += eventID; // we also need to check if we have all events - if (isDataComplete(mEventChecksum, info.maxEvents)) { + if (isDataComplete(mEventChecksum, maxEvents)) { LOG(info) << "ALL EVENTS HERE; CHECKSUM " << mEventChecksum; // flush remaining data and close file if (mMergerIOThread.joinable()) { mMergerIOThread.join(); } - mMergerIOThread = std::thread([info, this]() { mergingInProgress = true; mergeAndFlushData(); mergingInProgress = false; }); + mMergerIOThread = std::thread([this]() { mergingInProgress = true; mergeAndFlushData(); mergingInProgress = false; }); if (mMergerIOThread.joinable()) { mMergerIOThread.join(); } @@ -444,7 +448,7 @@ class O2HitMerger : public fair::mq::Device } if (mPipeToDriver != -1) { - if (write(mPipeToDriver, &info.eventID, sizeof(info.eventID)) == -1) { + if (write(mPipeToDriver, &eventID, sizeof(eventID)) == -1) { LOG(error) << "FAILED WRITING TO PIPE"; }; } @@ -452,9 +456,21 @@ class O2HitMerger : public fair::mq::Device return expectmore; } + // releases the buffered data of an event once it is flushed or discarded void cleanEvent(int eventID) { - // cleanup intermediate per-Event buffers + auto release = [eventID](auto& buffer) { + auto iter = buffer.find(eventID); + if (iter != buffer.end()) { + for (auto ptr : iter->second) { + delete ptr; + } + iter->second = {}; + } + }; + release(mMCTrackBuffer); + release(mTrackRefBuffer); + release(mSubEventInfoBuffer); } template @@ -558,11 +574,6 @@ class O2HitMerger : public fair::mq::Device channel.Send(reply); LOG(info) << "Forward publish MC tracks on channel"; } - - // cleanup buffered data - for (auto ptr : vectorOfSubEventMCTracks) { - delete ptr; // avoid this by using unique ptr - } } template @@ -609,11 +620,6 @@ class O2HitMerger : public fair::mq::Device targetbr->SetAddress(&dataaddr); targetbr->Fill(); targetbr->ResetAddress(); - - // cleanup mem - for (auto ptr : vectorOfT) { - delete ptr; // avoid this by using unique ptr - } } void updateTrackIdWithOffset(MCTrack& track, Int_t nprim, Int_t idelta0, Int_t idelta1) @@ -855,7 +861,7 @@ class O2HitMerger : public fair::mq::Device // intermediate structures to collect data per event std::thread mMergerIOThread; //! a thread used to do hit merging and IO flushing asynchronously - bool mergingInProgress = false; + std::atomic mergingInProgress{false}; Hashtable*>> mMCTrackBuffer; //! vector of sub-event track vectors; one per event Hashtable*>> mTrackRefBuffer; //! From 745cb1a1472a91316e539c7af9aaa351acf4a3bd Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 24 Sep 2026 21:09:46 +0200 Subject: [PATCH 6/7] Initialise the output pointers of the o2-sim hit merger This fixes the hit merger when it runs without disc output. - The kinematics and MC-header file and tree pointers were never initialised. - With --noDiscOutput they stayed garbage, and the merge tested them as if they were valid trees. - They are now initialised to nullptr. Co-Authored-By: Claude Opus 5.5 --- run/O2HitMerger.h | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/run/O2HitMerger.h b/run/O2HitMerger.h index 5814139c27cc1..6d795f2500ab0 100644 --- a/run/O2HitMerger.h +++ b/run/O2HitMerger.h @@ -849,10 +849,10 @@ class O2HitMerger : public fair::mq::Device std::string mOutFileName; //! // structures for the final flush - TFile* mOutFile; //! outfile for kinematics - TTree* mOutTree; //! tree (kinematics) associated to mOutFile - TFile* mMCHeaderOnlyOutFile; //! outfile for header only information - TTree* mMCHeaderTree; //! tree to hold MCHeader branch in mMCHeaderOnlyOutFile; + TFile* mOutFile = nullptr; //! outfile for kinematics + TTree* mOutTree = nullptr; //! tree (kinematics) associated to mOutFile + TFile* mMCHeaderOnlyOutFile = nullptr; //! outfile for header only information + TTree* mMCHeaderTree = nullptr; //! tree to hold MCHeader branch in mMCHeaderOnlyOutFile; template using Hashtable = tbb::concurrent_unordered_map; From 43df410930e18667377985285fba962d14918616 Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 24 Sep 2026 21:42:02 +0200 Subject: [PATCH 7/7] Finish detector events before sending hits in parallel o2-sim This fixes the order in which a parallel o2-sim worker finalizes and sends the hits of an event. - Detectors ran FinishEvent after SendData, while serial o2-sim (FairMCApplication) runs it before filling the output. - TRD sorts its hits in FinishEvent, and PHOS and CPV sort them and sum duplicates. - With TMessage transport their hits were written unsorted and, for PHOS and CPV, with duplicates. - With shared memory the merger read the buffers while the worker rewrote them, so TRD output differed between the two transports. - FinishEvent now runs for all detectors before SendData, and EndOfEvent after it. Co-Authored-By: Claude Opus 5.5 --- Steer/include/Steer/O2MCApplication.h | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/Steer/include/Steer/O2MCApplication.h b/Steer/include/Steer/O2MCApplication.h index 2ea1b9990a3f6..e43a61ec419b5 100644 --- a/Steer/include/Steer/O2MCApplication.h +++ b/Steer/include/Steer/O2MCApplication.h @@ -53,13 +53,17 @@ class O2MCApplication : public O2MCApplicationBase finishEventCommon(); + // detectors finalize their hits (e.g. sorting, summing duplicates) before these are sent + for (auto det : listActiveDetectors) { + det->FinishEvent(); + } + // This special finish event version does not fill the output tree of FairRootManager // but forwards the data to the HitMerger SendData(); // call end of event on active detectors for (auto det : listActiveDetectors) { - det->FinishEvent(); det->EndOfEvent(); } fStack->Reset();