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(); diff --git a/run/O2HitMerger.h b/run/O2HitMerger.h index 15f58c6dba351..6d795f2500ab0 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) @@ -654,34 +660,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 +712,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 +828,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); @@ -858,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; @@ -870,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; //! diff --git a/run/O2PrimaryServerDevice.h b/run/O2PrimaryServerDevice.h index b8703ffcddb28..d5608078593d4 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(); @@ -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)); 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; }