Stability fixes for o2-sim

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

run/O2HitMerger.h

@@ -67,6 +67,7 @@
 #include <list>
 #include <csignal>
 #include <mutex>
+#include <atomic>
 #include <filesystem>
 #include <functional>
 
@@ -399,6 +400,9 @@ class O2HitMerger : public fair::mq::Device
     int index = 0;
     auto infoptr = o2::base::decodeTMessage<o2::data::SubEventInfo*>(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<uint32_t, uint32_t>(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<uint32_t>(mEventChecksum, info.maxEvents)) {
+      if (isDataComplete<uint32_t>(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,17 +448,29 @@ 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";
         };
       }
     }
     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 <typename T>
@@ -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 <typename T, typename M>
@@ -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 <class K, class V>
   using Hashtable = tbb::concurrent_unordered_map<K, V>;
@@ -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<bool> mergingInProgress{false};
 
   Hashtable<int, std::vector<std::vector<o2::MCTrack>*>> mMCTrackBuffer;         //! vector of sub-event track vectors; one per event
   Hashtable<int, std::vector<std::vector<o2::TrackReference>*>> mTrackRefBuffer; //!

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

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