DPL: move away from MessageSet::header / payload

Abstract header / payload retrieval, with the idea that get_header / get_payload
will work on any range of fair::mq::MessagePtrs.

For now we only do the first header / payload pair only, to validate the trivial change.

Author: ktf

Framework/Core/include/Framework/DataModelViews.h

@@ -153,7 +153,7 @@ struct get_header {
   // ends the pipeline, returns the number of parts
   template <typename R>
     requires std::ranges::random_access_range<R> && std::ranges::sized_range<R>
-  friend fair::mq::MessagePtr& operator|(R&& r, get_header self)
+  friend auto& operator|(R&& r, get_header self)
   {
     return r[(r | get_dataref_indices{self.id, 0}).headerIdx];
   }
@@ -165,7 +165,7 @@ struct get_payload {
   // ends the pipeline, returns the number of parts
   template <typename R>
     requires std::ranges::random_access_range<R> && std::ranges::sized_range<R>
-  friend fair::mq::MessagePtr& operator|(R&& r, get_payload self)
+  friend auto& operator|(R&& r, get_payload self)
   {
     return r[(r | get_dataref_indices{self.part, self.subPart}).payloadIdx];
   }

Framework/Core/src/DataProcessingDevice.cxx

@@ -2153,7 +2153,7 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
       return currentSetOfInputs[i].getNumberOfPairs();
     };
     auto refCountGetter = [&currentSetOfInputs](size_t idx) -> int {
-      auto& header = static_cast<const fair::mq::shmem::Message&>(*currentSetOfInputs[idx].header(0));
+      auto& header = static_cast<const fair::mq::shmem::Message&>(*(currentSetOfInputs[idx].messages | get_header{0}));
       return header.GetRefCount();
     };
     return InputSpan{getter, nofPartsGetter, refCountGetter, currentSetOfInputs.size()};

Framework/Core/src/DataRelayer.cxx

@@ -184,11 +184,11 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
       // We check that no data is already there for the given cell
       // it is enough to check the first element
       auto& part = mCache[ti * mDistinctRoutesIndex.size() + expirator.routeIndex.value];
-      if (!part.messages.empty() && part.header(0) != nullptr) {
+      if (!part.messages.empty() && (part.messages | get_header{0}) != nullptr) {
         headerPresent++;
         continue;
       }
-      if (!part.messages.empty() && part.payload(0) != nullptr) {
+      if (!part.messages.empty() && (part.messages | get_payload{0, 0}) != nullptr) {
         payloadPresent++;
         continue;
       }
@@ -227,7 +227,7 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
         return partial[idx].messages | count_parts{};
       };
       auto refCountGetter = [&partial](size_t idx) -> int {
-        auto& header = static_cast<const fair::mq::shmem::Message&>(*partial[idx].header(0));
+        auto& header = static_cast<const fair::mq::shmem::Message&>(*(partial[idx].messages | get_header{0}));
         return header.GetRefCount();
       };
       InputSpan span{getter, nPartsGetter, refCountGetter, static_cast<size_t>(partial.size())};
@@ -246,8 +246,8 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
       activity.expiredSlots++;
 
       mTimesliceIndex.markAsDirty(slot, true);
-      assert(part.header(0) != nullptr);
-      assert(part.payload(0) != nullptr);
+      assert((part.messages | get_header{0}) != nullptr);
+      assert((part.messages | get_payload{0, 0}) != nullptr);
     }
   }
   LOGP(debug, "DataRelayer::processDanglingInputs headerPresent:{}, payloadPresent:{}, noCheckers:{}, badSlot:{}, checkerDenied:{}",
@@ -800,7 +800,7 @@ void DataRelayer::getReadyToProcess(std::vector<DataRelayer::RecordAction>& comp
       return partial[idx].messages | count_parts{};
     };
     auto refCountGetter = [&partial](size_t idx) -> int {
-      auto& header = static_cast<const fair::mq::shmem::Message&>(*partial[idx].header(0));
+      auto& header = static_cast<const fair::mq::shmem::Message&>(*(partial[idx].messages | get_header{0}));
       return header.GetRefCount();
     };
     InputSpan span{getter, nPartsGetter, refCountGetter, static_cast<size_t>(partial.size())};

Framework/Core/test/test_DataRelayer.cxx

@@ -798,11 +798,11 @@ TEST_CASE("DataRelayer")
     // one message set containing number of added sequences of messages
     REQUIRE((messageSet[0].messages | count_parts{}) == sequenceSize.size());
     size_t counter = 0;
-    for (auto seqid = 0; seqid < sequenceSize.size(); ++seqid) {
+    for (size_t seqid = 0; seqid < sequenceSize.size(); ++seqid) {
       REQUIRE(messageSet[0].getNumberOfPayloads(seqid) == sequenceSize[seqid]);
-      for (auto pi = 0; pi < messageSet[0].getNumberOfPayloads(seqid); ++pi) {
-        REQUIRE(messageSet[0].payload(seqid, pi));
-        auto const* data = messageSet[0].payload(seqid, pi)->GetData();
+      for (size_t pi = 0; pi < messageSet[0].getNumberOfPayloads(seqid); ++pi) {
+        REQUIRE((messageSet[0].messages | get_payload{seqid, pi}));
+        auto const* data = (messageSet[0].messages | get_payload{seqid, pi})->GetData();
         REQUIRE(*(reinterpret_cast<size_t const*>(data)) == counter);
         ++counter;
       }