Send the hit transport mode with the hits in o2-sim

This makes the hit merger decode each detector's hits the way the worker sent them.

- Worker and merger each read the global ShmManager::isOperational() to choose between shared memory and TMessage.
- That flag lives in the segment and flips when any late worker fails to attach, so hits already in flight could be decoded the wrong way.
- The per-detector header message is now a HitsHeader carrying the DetID and the mode, decided once per detector in attachHits.
- collectHits takes the mode from the header.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: sawenzel

Detectors/Base/include/DetectorsBase/Detector.h

@@ -187,7 +187,7 @@ class Detector : public FairDetector
   // and to decode it
   virtual void attachHits(fair::mq::Channel&, fair::mq::Parts&) = 0;
   virtual void fillHitBranch(TTree& tr, fair::mq::Parts& parts, int& index) = 0;
-  virtual void collectHits(int eventID, fair::mq::Parts& parts, int& index) = 0;
+  virtual void collectHits(int eventID, fair::mq::Parts& parts, int& index, bool shm) = 0;
   virtual void mergeHitEntriesAndFlush(int eventID,
                                        TTree& target,
                                        std::vector<int> const& trackoffsets,
@@ -298,7 +298,13 @@ T decodeTMessage(fair::mq::Parts& dataparts, int index)
   return static_cast<T>(decodeTMessageCore(dataparts, index));
 }
 
-void attachDetIDHeaderMessage(int id, fair::mq::Channel& channel, fair::mq::Parts& parts);
+// header message preceding the hits of one detector
+struct HitsHeader {
+  int detID;
+  bool shm; // whether the hits follow as shared-memory references or as TMessages
+};
+
+void attachHitsHeaderMessage(HitsHeader const& header, fair::mq::Channel& channel, fair::mq::Parts& parts);
 
 template <typename T>
 TBranch* getOrMakeBranch(TTree& tree, const char* brname, T* ptr)
@@ -360,10 +366,12 @@ class DetImpl : public o2::base::Detector
       return;
     }
 
-    attachDetIDHeaderMessage(GetDetId(), channel, parts); // the DetId s are universal as they come from o2::detector::DetID
+    // decide the transport once, so that the header and all hit messages agree
+    const bool shm = UseShm<Det>::value && o2::utils::ShmManager::Instance().isOperational();
+    attachHitsHeaderMessage({GetDetId(), shm}, channel, parts); // the DetId s are universal as they come from o2::detector::DetID
 
     while (auto hits = static_cast<Det*>(this)->Det::getHits(probe++)) {
-      if (!UseShm<Det>::value || !o2::utils::ShmManager::Instance().isOperational()) {
+      if (!shm) {
         attachTMessage(*hits, channel, parts);
       } else {
         // this is the shared mem variant
@@ -542,7 +550,7 @@ class DetImpl : public o2::base::Detector
   /// Collect Hits available as incoming message (shared mem or not)
   /// inside this process for later streaming to output. A function needed
   /// by the hit-merger process (not for direct use by users)
-  void collectHits(int eventID, fair::mq::Parts& parts, int& index) override
+  void collectHits(int eventID, fair::mq::Parts& parts, int& index, bool shm) override
   {
     using Hit_t = typename std::remove_pointer<decltype(static_cast<Det*>(this)->Det::getHits(0))>::type;
     using Collector_t = tbb::concurrent_unordered_map<int, std::vector<std::vector<std::unique_ptr<Hit_t>>>>;
@@ -582,7 +590,7 @@ class DetImpl : public o2::base::Detector
     };
 
     while (name.size() > 0) {
-      if (!UseShm<Det>::value || !o2::utils::ShmManager::Instance().isOperational()) {
+      if (!shm) {
         // for each branch name we extract/decode hits from the message parts ...
         auto hitsptr = decodeTMessage<HitPtr_t>(parts, index++);
         if (hitsptr) {

Detectors/Base/src/Detector.cxx

@@ -215,9 +215,9 @@ void attachMessageBufferToParts(fair::mq::Parts& parts, fair::mq::Channel& chann
   o2::framework::TMessageSerializer::serialize(buffer, data, cl);
   parts.AddPart(std::move(msg));
 }
-void attachDetIDHeaderMessage(int id, fair::mq::Channel& channel, fair::mq::Parts& parts)
+void attachHitsHeaderMessage(HitsHeader const& header, fair::mq::Channel& channel, fair::mq::Parts& parts)
 {
-  std::unique_ptr<fair::mq::Message> message(channel.NewSimpleMessage(id));
+  std::unique_ptr<fair::mq::Message> message(channel.NewSimpleMessage(header));
   parts.AddPart(std::move(message));
 }
 void attachShmMessage(void* hits_ptr, fair::mq::Channel& channel, fair::mq::Parts& parts, ShmBusyFlag* busy_ptr)

Detectors/Base/test/testStack.cxx

@@ -117,7 +117,7 @@ class TestDetector : public o2::base::Detector
   std::string getHitBranchNames(int) const override { return {}; }
   void attachHits(fair::mq::Channel&, fair::mq::Parts&) override {}
   void fillHitBranch(TTree&, fair::mq::Parts&, int&) override {}
-  void collectHits(int, fair::mq::Parts&, int&) override {}
+  void collectHits(int, fair::mq::Parts&, int&, bool) override {}
   void mergeHitEntriesAndFlush(int, TTree&, std::vector<int> const&, std::vector<int> const&,
                                std::vector<int> const&) override {}
   void mergeHitEntries(TTree&, TTree&, std::vector<int> const&, std::vector<int> const&,

run/O2HitMerger.h

@@ -297,18 +297,18 @@ class O2HitMerger : public fair::mq::Device
 
   void consumeHits(int eventID, fair::mq::Parts& data, int& index)
   {
-    auto detIDmessage = std::move(data.At(index++));
-    // this should be a detector ID
-    if (detIDmessage->GetSize() == 4) {
-      auto ptr = (int*)detIDmessage->GetData();
-      o2::detectors::DetID id(ptr[0]);
-      LOG(debug2) << "I1 " << ptr[0] << " NAME " << id.getName() << " MB "
+    auto headermessage = std::move(data.At(index++));
+    // this should be the header announcing the hits of one detector
+    if (headermessage->GetSize() == sizeof(o2::base::HitsHeader)) {
+      auto header = *static_cast<o2::base::HitsHeader const*>(headermessage->GetData());
+      o2::detectors::DetID id(header.detID);
+      LOG(debug2) << "I1 " << header.detID << " NAME " << id.getName() << " MB "
                   << data.At(index)->GetSize() / 1024. / 1024.;
 
       // get the detector that can interpret it
       auto detector = mDetectorInstances[id].get();
       if (detector) {
-        detector->collectHits(eventID, data, index);
+        detector->collectHits(eventID, data, index, header.shm);
       }
     }
   }