fixup! Generalizing DataRelayer to relay an array of messages

Author: matthiasrichter

Framework/Core/include/Framework/DataRelayer.h

@@ -91,14 +91,15 @@ class DataRelayer
   /// @a rawHeader raw header pointer
   /// @a messages pointer to array of messages
   /// @a nMessages size of the array
-  /// @a createPairs create PartRef objects with header-payload pairs if true
-  ///                or one PartRef object with multiple payloads
-  /// is the header which is common across all subsequent elements.
+  /// @a nPayloads number of payploads in the message sequence, default is 1
+  ///              which is the standard header-payload message pair, in this
+  ///              case nMessages / 2 pairs will be inserted and considered
+  ///              separate parts
   /// Notice that we expect that the header is an O2 Header Stack
   RelayChoice relay(void const* rawHeader,
                     std::unique_ptr<FairMQMessage>* messages,
                     size_t nMessages,
-                    bool createPairs = true);
+                    size_t nPayloads = 1);
 
   /// @returns the actions ready to be performed.
   void getReadyToProcess(std::vector<RecordAction>& completed);

Framework/Core/src/DataRelayer.cxx

@@ -204,7 +204,7 @@ DataRelayer::RelayChoice
   DataRelayer::relay(void const* rawHeader,
                      std::unique_ptr<FairMQMessage>* messages,
                      size_t nMessages,
-                     bool createPairs)
+                     size_t nPayloads)
 {
   std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);
   DataProcessingHeader const* dph = o2::header::get<DataProcessingHeader*>(rawHeader);
@@ -280,7 +280,7 @@ DataRelayer::RelayChoice
   auto saveInSlot = [&cachedStateMetrics = mCachedStateMetrics,
                      &messages,
                      &nMessages,
-                     &createPairs,
+                     &nPayloads,
                      &cache,
                      &numInputTypes,
                      &metrics](TimesliceId timeslice, int input, TimesliceSlot slot) {
@@ -289,13 +289,11 @@ DataRelayer::RelayChoice
     cachedStateMetrics[cacheIdx] = CacheEntryStatus::PENDING;
     // TODO: make sure that multiple parts can only be added within the same call of
     // DataRelayer::relay
-    // depending on message model, add pairs or sequences with variable number of payloads
-    if (createPairs) {
-      for (size_t pi = 0; pi < nMessages / 2; ++pi) {
-        target.add([&messages, &pi](size_t i) -> FairMQMessagePtr& {return messages[pi * 2 + i];}, 2);
-      }
-    } else {
-      target.add([&messages](size_t i) -> FairMQMessagePtr& {return messages[i];}, nMessages);
+    assert(nPayloads > 0);
+    for (size_t mi = 0; mi < nMessages; ++mi) {
+      assert(mi + nPayloads < nMessages);
+      target.add([&messages, &mi](size_t i) -> FairMQMessagePtr& {return messages[mi + i];}, nPayloads + 1);
+      mi += nPayloads;
     }
   };
 
@@ -516,8 +514,6 @@ void DataRelayer::getReadyToProcess(std::vector<DataRelayer::RecordAction>& comp
       return DataRef{};
     };
     auto nPartsGetter = [&partial](size_t idx) {
-      // with PartRef holding potentially  multiple payload messages,
-      // one would need to accumulate the result, think about caching
       return partial[idx].size();
     };
     InputSpan span{getter, nPartsGetter, static_cast<size_t>(partial.size())};

Framework/Core/test/test_DataRelayer.cxx

@@ -698,16 +698,17 @@ BOOST_AUTO_TEST_CASE(SplitPayloadSequence)
 
   for (size_t i = 0; i < nSplitParts; ++i) {
     splitParts.emplace_back(transport->CreateMessage(100));
+    *(reinterpret_cast<size_t*>(splitParts.back()->GetData())) = i;
   }
   BOOST_REQUIRE_EQUAL(splitParts.size(), nSplitParts + 1);
 
-  relayer.relay(splitParts[0]->GetData(), splitParts.data(), splitParts.size(), false);
+  relayer.relay(splitParts[0]->GetData(), splitParts.data(), splitParts.size(), nSplitParts);
   std::vector<RecordAction> ready;
   relayer.getReadyToProcess(ready);
   BOOST_REQUIRE_EQUAL(ready.size(), 1);
   BOOST_REQUIRE_EQUAL(ready[0].op, CompletionPolicy::CompletionOp::Consume);
   auto messageSet = relayer.getInputsForTimeslice(ready[0].slot);
-  // we have one input route and thus one message set containing one PartRef
+  // we have one input route and thus one message set containing one sequence of messages
   BOOST_REQUIRE_EQUAL(messageSet.size(), 1);
   BOOST_REQUIRE_EQUAL(messageSet[0].size(), 1);
   BOOST_CHECK_EQUAL(messageSet[0].getNumberOfPayloads(0), 100);