DPL: navigate a MessageSet without caching pairs

Author: ktf

Framework/Core/include/Framework/MessageSet.h

@@ -12,13 +12,13 @@
 #define FRAMEWORK_MESSAGESET_H
 
 #include "Framework/PartRef.h"
+#include <fairmq/Message.h>
+#include "Framework/DataModelViews.h"
 #include <memory>
 #include <vector>
 #include <cassert>
 
-namespace o2
-{
-namespace framework
+namespace o2::framework
 {
 
 /// A set of inflight messages.
@@ -31,41 +31,23 @@ namespace framework
 /// O2 message model. For this purpose, also the pair index is filled and can
 /// be used to access header and payload associated with a pair
 struct MessageSet {
-  struct Index {
-    Index(size_t p, size_t s) : position(p), size(s) {}
-    size_t position = 0;
-    size_t size = 0;
-  };
   // linear storage of messages
   std::vector<fair::mq::MessagePtr> messages;
-  // message map describes O2 messages consisting of a header message and
-  // payload message(s), index describes position in the linear storage
-  std::vector<Index> messageMap;
-  // pair map describes all messages in one sequence of header-payload pairs and
-  // where in the message index the associated header and payload can be found
-  struct PairMapping {
-    PairMapping(size_t partId, size_t payloadId) : partIndex(partId), payloadIndex(payloadId) {}
-    // O2 message where the pair is located in
-    size_t partIndex = 0;
-    // payload index within the O2 message
-    size_t payloadIndex = 0;
-  };
-  std::vector<PairMapping> pairMap;
 
   MessageSet()
-    : messages(), messageMap(), pairMap()
+    : messages()
   {
   }
 
   template <typename F>
   MessageSet(F getter, size_t size)
-    : messages(), messageMap(), pairMap()
+    : messages()
   {
     add(std::forward<F>(getter), size);
   }
 
   MessageSet(MessageSet&& other)
-    : messages(std::move(other.messages)), messageMap(std::move(other.messageMap)), pairMap(std::move(other.pairMap))
+    : messages(std::move(other.messages))
   {
     other.clear();
   }
@@ -76,36 +58,32 @@ struct MessageSet {
       return *this;
     }
     messages = std::move(other.messages);
-    messageMap = std::move(other.messageMap);
-    pairMap = std::move(other.pairMap);
     other.clear();
     return *this;
   }
 
   /// get number of in-flight O2 messages
-  size_t size() const
+  [[nodiscard]] size_t size() const
   {
-    return messageMap.size();
+    return messages | count_parts{};
   }
 
   /// get number of header-payload pairs
-  size_t getNumberOfPairs() const
+  [[nodiscard]] size_t getNumberOfPairs() const
   {
-    return pairMap.size();
+    return messages | count_payloads{};
   }
 
   /// get number of payloads for an in-flight message
-  size_t getNumberOfPayloads(size_t mi) const
+  [[nodiscard]] size_t getNumberOfPayloads(size_t mi) const
   {
-    return messageMap[mi].size;
+    return messages | get_num_payloads{mi};
   }
 
   /// clear the set
   void clear()
   {
     messages.clear();
-    messageMap.clear();
-    pairMap.clear();
   }
 
   // this is more or less legacy
@@ -122,8 +100,6 @@ struct MessageSet {
   // add  content of the part ref
   void add(PartRef&& ref)
   {
-    pairMap.emplace_back(messageMap.size(), 0);
-    messageMap.emplace_back(messages.size(), 1);
     messages.emplace_back(std::move(ref.header));
     messages.emplace_back(std::move(ref.payload));
   }
@@ -132,53 +108,32 @@ struct MessageSet {
   template <typename F>
   void add(F getter, size_t size)
   {
-    auto partid = messageMap.size();
-    messageMap.emplace_back(messages.size(), size - 1);
     for (size_t i = 0; i < size; ++i) {
-      if (i > 0) {
-        pairMap.emplace_back(partid, i - 1);
-      }
       messages.emplace_back(std::move(getter(i)));
     }
   }
 
   fair::mq::MessagePtr& header(size_t partIndex)
   {
-    return messages[messageMap[partIndex].position];
+    return messages[(messages | get_dataref_indices{partIndex, 0}).headerIdx];
   }
 
   fair::mq::MessagePtr& payload(size_t partIndex, size_t payloadIndex = 0)
   {
-    assert(partIndex < messageMap.size());
-    assert(messageMap[partIndex].position + payloadIndex + 1 < messages.size());
-    return messages[messageMap[partIndex].position + payloadIndex + 1];
-  }
-
-  fair::mq::MessagePtr const& header(size_t partIndex) const
-  {
-    return messages[messageMap[partIndex].position];
+    return messages[(messages | get_dataref_indices{partIndex, payloadIndex}).payloadIdx];
   }
 
-  fair::mq::MessagePtr const& payload(size_t partIndex, size_t payloadIndex = 0) const
+  [[nodiscard]] fair::mq::MessagePtr const& header(size_t partIndex) const
   {
-    assert(partIndex < messageMap.size());
-    assert(messageMap[partIndex].position + payloadIndex + 1 < messages.size());
-    return messages[messageMap[partIndex].position + payloadIndex + 1];
+    return messages[(messages | get_dataref_indices{partIndex, 0}).headerIdx];
   }
 
-  fair::mq::MessagePtr const& associatedHeader(size_t pos) const
+  [[nodiscard]] fair::mq::MessagePtr const& payload(size_t partIndex, size_t payloadIndex = 0) const
   {
-    return messages[messageMap[pairMap[pos].partIndex].position];
-  }
-
-  fair::mq::MessagePtr const& associatedPayload(size_t pos) const
-  {
-    auto partIndex = pairMap[pos].partIndex;
-    auto payloadIndex = pairMap[pos].payloadIndex;
-    return messages[messageMap[partIndex].position + payloadIndex + 1];
+    return messages[(messages | get_dataref_indices{partIndex, payloadIndex}).payloadIdx];
   }
 };
 
-} // namespace framework
-} // namespace o2
+} // namespace o2::framework
+
 #endif // FRAMEWORK_MESSAGESET_H

Framework/Core/src/DataProcessingDevice.cxx

@@ -2140,8 +2140,10 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
         //   sequence is the header message
         // - each part has one or more payload messages
         // - InputRecord provides all payloads as header-payload pair
-        auto const& headerMsg = currentSetOfInputs[i].associatedHeader(partindex);
-        auto const& payloadMsg = currentSetOfInputs[i].associatedPayload(partindex);
+        //
+        auto const& [headerMsgIndex, payloadMsgIndex] = currentSetOfInputs[i].messages | get_pair{partindex};
+        auto const& headerMsg = currentSetOfInputs[i].messages[headerMsgIndex];
+        auto const& payloadMsg = currentSetOfInputs[i].messages[payloadMsgIndex];
         headerptr = static_cast<char const*>(headerMsg->GetData());
         payloadptr = payloadMsg ? static_cast<char const*>(payloadMsg->GetData()) : nullptr;
         payloadSize = payloadMsg ? payloadMsg->GetSize() : 0;